car-server-core 0.55.0

//! Operator guidance delivered at native model-turn boundaries.
//! The session snapshot owns durability; this inbox only coordinates the live loop.

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

use super::session::{CoderEventKind, EventSink};

#[derive(Debug, Default)]
struct Pending {
    open: bool,
    messages: VecDeque<String>,
}

#[derive(Debug, Default)]
pub struct SteeringInbox(Mutex<Pending>);

impl SteeringInbox {
    pub fn is_open(&self) -> bool {
        self.0.lock().expect("steering inbox poisoned").open
    }

    pub fn has_pending(&self) -> bool {
        !self
            .0
            .lock()
            .expect("steering inbox poisoned")
            .messages
            .is_empty()
    }
    pub fn enter<'a>(self: &Arc<Self>, sink: &'a EventSink) -> SteeringScope<'a> {
        self.0.lock().expect("steering inbox poisoned").open = true;
        SteeringScope {
            inbox: self.clone(),
            sink,
        }
    }

    /// Persist before acknowledgement. Holding the inbox lock makes admission
    /// atomic with loop completion, and a failed write never queues guidance.
    pub fn enqueue(
        &self,
        text: String,
        persist: impl FnOnce() -> Result<(), String>,
    ) -> Result<(), String> {
        let mut pending = self.0.lock().expect("steering inbox poisoned");
        if !pending.open {
            return Err("This task is not accepting steering. Native tasks accept guidance while running; use a follow-up conversation after the task finishes.".into());
        }
        if pending.messages.len() >= 16 {
            return Err("Too many queued instructions. Wait for the agent to receive them before sending more.".into());
        }
        persist()?;
        pending.messages.push_back(text);
        Ok(())
    }

    pub fn take(&self) -> Vec<String> {
        self.0
            .lock()
            .expect("steering inbox poisoned")
            .messages
            .drain(..)
            .collect()
    }
}

pub struct SteeringScope<'a> {
    inbox: Arc<SteeringInbox>,
    sink: &'a EventSink,
}

impl Drop for SteeringScope<'_> {
    fn drop(&mut self) {
        let mut pending = self.inbox.0.lock().expect("steering inbox poisoned");
        pending.open = false;
        for text in pending.messages.drain(..) {
            self.sink.emit(CoderEventKind::OperatorGuidance {
                text,
                status: "not_applied".into(),
            });
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn steering_requires_live_scope_and_durable_acceptance() {
        let inbox = Arc::new(SteeringInbox::default());
        assert!(inbox.enqueue("before".into(), || Ok(())).is_err());
        let (sink, events) = EventSink::collecting("steering");
        let scope = inbox.enter(&sink);
        assert!(inbox
            .enqueue("lost write".into(), || Err("disk full".into()))
            .is_err());
        assert!(inbox.take().is_empty());
        inbox
            .enqueue("retain unfinished work".into(), || Ok(()))
            .unwrap();
        drop(scope);
        assert!(!inbox.is_open());
        assert!(inbox.enqueue("too late".into(), || Ok(())).is_err());
        assert!(events.lock().unwrap().iter().any(|event| matches!(&event.kind,
            CoderEventKind::OperatorGuidance { text, status } if text == "retain unfinished work" && status == "not_applied")));
    }
}