prospero_core/caliband/
mod.rs

1//! Caliban integration: wire types, NDJSON framing, the control client, and
2//! the stream-json normalizer. The wire format is the only coupling to caliban.
3
4pub mod client;
5pub mod sources;
6pub mod stream;
7pub mod transport;
8pub mod wire;
9
10use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
11
12use crate::error::{CoreError, Result};
13
14/// Write one JSON value as an NDJSON frame (compact JSON + `\n`).
15pub(crate) async fn write_frame<W, T>(w: &mut W, value: &T) -> Result<()>
16where
17    W: AsyncWriteExt + Unpin,
18    T: serde::Serialize,
19{
20    let mut line = serde_json::to_vec(value)?;
21    line.push(b'\n');
22    w.write_all(&line).await?;
23    w.flush().await?;
24    Ok(())
25}
26
27/// Read exactly one NDJSON frame and deserialize it. Returns a protocol error
28/// if the stream ends before a full line is read.
29pub(crate) async fn read_frame<R, T>(r: &mut R) -> Result<T>
30where
31    R: AsyncBufReadExt + Unpin,
32    T: serde::de::DeserializeOwned,
33{
34    let mut line = String::new();
35    let n = r.read_line(&mut line).await?;
36    if n == 0 {
37        return Err(CoreError::Protocol(
38            "connection closed before a reply frame was read".into(),
39        ));
40    }
41    let value = serde_json::from_str(line.trim_end())?;
42    Ok(value)
43}
44
45#[cfg(test)]
46mod tests {
47    use super::*;
48    use crate::caliband::wire::CtlRequest;
49    use tokio::io::BufReader;
50
51    #[tokio::test]
52    async fn frame_round_trips_through_a_pipe() {
53        // Write a frame into an in-memory buffer, then read it back.
54        let mut buf: Vec<u8> = Vec::new();
55        write_frame(&mut buf, &CtlRequest::List).await.unwrap();
56        assert_eq!(buf, b"{\"kind\":\"list\"}\n");
57
58        let mut reader = BufReader::new(&buf[..]);
59        let req: CtlRequest = read_frame(&mut reader).await.unwrap();
60        assert_eq!(req, CtlRequest::List);
61    }
62
63    #[tokio::test]
64    async fn read_frame_errors_on_empty_stream() {
65        let empty: &[u8] = b"";
66        let mut reader = BufReader::new(empty);
67        let r: Result<CtlRequest> = read_frame(&mut reader).await;
68        assert!(matches!(r, Err(CoreError::Protocol(_))));
69    }
70}
prospero_core/caliband/mod.rs

prospero_core/caliband/
mod.rs
