bamboo_plugin_protocol/
publisher.rs

1use std::collections::VecDeque;
2use std::sync::{Mutex, TryLockError};
3
4use thiserror::Error;
5
6use crate::{ToolEventBuildError, ToolEventV1};
7
8/// Non-blocking injection seam owned by one Bamboo runtime/AppState.
9///
10/// Implementations MUST return immediately and MUST NOT perform process I/O.
11/// Runtime routing and queues are intentionally outside this protocol slice.
12pub trait ToolEventPublisher: Send + Sync {
13    /// Fast capability hint so the default no-op path performs no event DTO
14    /// allocation. Implementations should keep the default `true`.
15    fn is_enabled(&self) -> bool {
16        true
17    }
18
19    fn try_publish(&self, event: ToolEventV1) -> Result<(), ToolEventPublishError>;
20}
21
22/// Default publisher used by the SDK, server, and tests unless explicitly
23/// injected. It preserves historical behavior and allocations at the sink.
24#[derive(Clone, Copy, Debug, Default)]
25pub struct NoopToolEventPublisher;
26
27impl ToolEventPublisher for NoopToolEventPublisher {
28    fn is_enabled(&self) -> bool {
29        false
30    }
31
32    fn try_publish(&self, _event: ToolEventV1) -> Result<(), ToolEventPublishError> {
33        Ok(())
34    }
35}
36
37/// Bounded in-memory recorder for tests and embedders.
38///
39/// Both publication and observation use `Mutex::try_lock`; contention is a
40/// drop/error signal and can never park the executing tool thread.
41#[derive(Debug)]
42pub struct InMemoryToolEventRecorder {
43    capacity: usize,
44    events: Mutex<VecDeque<ToolEventV1>>,
45}
46
47impl InMemoryToolEventRecorder {
48    pub fn new(capacity: usize) -> Result<Self, ToolEventPublishError> {
49        if capacity == 0 {
50            return Err(ToolEventPublishError::InvalidCapacity);
51        }
52        Ok(Self {
53            capacity,
54            // Capacity is a logical bound. Avoid eagerly allocating a caller-
55            // supplied capacity before the first bounded event arrives.
56            events: Mutex::new(VecDeque::new()),
57        })
58    }
59
60    pub fn try_snapshot(&self) -> Result<Vec<ToolEventV1>, ToolEventPublishError> {
61        let events = self.try_lock()?;
62        Ok(events.iter().cloned().collect())
63    }
64
65    pub fn try_drain(&self) -> Result<Vec<ToolEventV1>, ToolEventPublishError> {
66        let mut events = self.try_lock()?;
67        Ok(events.drain(..).collect())
68    }
69
70    fn try_lock(
71        &self,
72    ) -> Result<std::sync::MutexGuard<'_, VecDeque<ToolEventV1>>, ToolEventPublishError> {
73        match self.events.try_lock() {
74            Ok(events) => Ok(events),
75            Err(TryLockError::WouldBlock) => Err(ToolEventPublishError::Busy),
76            Err(TryLockError::Poisoned(_)) => Err(ToolEventPublishError::Poisoned),
77        }
78    }
79}
80
81impl ToolEventPublisher for InMemoryToolEventRecorder {
82    fn try_publish(&self, event: ToolEventV1) -> Result<(), ToolEventPublishError> {
83        event
84            .validate_bounds()
85            .map_err(ToolEventPublishError::InvalidEvent)?;
86        let mut events = self.try_lock()?;
87        if events.len() >= self.capacity {
88            return Err(ToolEventPublishError::Full {
89                capacity: self.capacity,
90            });
91        }
92        events.push_back(event);
93        Ok(())
94    }
95}
96
97#[derive(Clone, Debug, Error, PartialEq, Eq)]
98pub enum ToolEventPublishError {
99    #[error("tool event recorder capacity must be greater than zero")]
100    InvalidCapacity,
101    #[error("tool event publisher is busy")]
102    Busy,
103    #[error("tool event publisher is full (capacity {capacity})")]
104    Full { capacity: usize },
105    #[error("tool event publisher state is poisoned")]
106    Poisoned,
107    #[error("invalid tool event: {0}")]
108    InvalidEvent(ToolEventBuildError),
109    #[error("tool event publisher failed: {0}")]
110    Failed(String),
111}
112
113#[cfg(test)]
114mod tests {
115    use crate::{FileChangedV1, ToolEventContextV1};
116
117    use super::*;
118
119    fn event(call_id: &str) -> ToolEventV1 {
120        ToolEventV1::file_changed(
121            ToolEventContextV1::bounded("session", "root-session", "Write", call_id).unwrap(),
122            FileChangedV1::bounded("/root/file.rs").unwrap(),
123        )
124        .unwrap()
125    }
126
127    #[test]
128    fn recorder_is_bounded() {
129        let recorder = InMemoryToolEventRecorder::new(1).unwrap();
130        recorder.try_publish(event("one")).unwrap();
131        assert_eq!(
132            recorder.try_publish(event("two")),
133            Err(ToolEventPublishError::Full { capacity: 1 })
134        );
135        assert_eq!(recorder.try_snapshot().unwrap().len(), 1);
136    }
137
138    #[test]
139    fn contended_recorder_returns_without_waiting() {
140        let recorder = InMemoryToolEventRecorder::new(1).unwrap();
141        let _guard = recorder.events.lock().unwrap();
142        assert_eq!(
143            recorder.try_publish(event("busy")),
144            Err(ToolEventPublishError::Busy)
145        );
146    }
147
148    #[test]
149    fn poisoned_recorder_returns_an_explicit_error() {
150        let recorder = std::sync::Arc::new(InMemoryToolEventRecorder::new(1).unwrap());
151        let poison_target = recorder.clone();
152        let poisoned = std::thread::spawn(move || {
153            let _guard = poison_target.events.lock().unwrap();
154            panic!("poison recorder for deterministic coverage");
155        });
156        assert!(poisoned.join().is_err());
157
158        assert_eq!(
159            recorder.try_publish(event("poisoned")),
160            Err(ToolEventPublishError::Poisoned)
161        );
162    }
163}
bamboo_plugin_protocol/publisher.rs

bamboo_plugin_protocol/
publisher.rs
