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 serde_json::json;
50    use tokio::io::BufReader;
51
52    #[tokio::test]
53    async fn frame_round_trips_through_a_pipe() {
54        // Write a frame into an in-memory buffer, then read it back.
55        let mut buf: Vec<u8> = Vec::new();
56        write_frame(&mut buf, &CtlRequest::List).await.unwrap();
57        assert_eq!(buf, b"{\"kind\":\"list\"}\n");
58
59        let mut reader = BufReader::new(&buf[..]);
60        let req: CtlRequest = read_frame(&mut reader).await.unwrap();
61        // The contract's enums derive no PartialEq; the wire form is the contract.
62        assert_eq!(serde_json::to_value(&req).unwrap(), json!({"kind": "list"}));
63    }
64
65    #[tokio::test]
66    async fn read_frame_errors_on_empty_stream() {
67        let empty: &[u8] = b"";
68        let mut reader = BufReader::new(empty);
69        let r: Result<CtlRequest> = read_frame(&mut reader).await;
70        assert!(matches!(r, Err(CoreError::Protocol(_))));
71    }
72}
prospero_core/caliband/mod.rs

prospero_core/caliband/
mod.rs
