agent_sdk_toolkit/protocol/
line_transport.rs

1//! Newline-delimited JSON-RPC transport helpers. Use this module for deterministic
2//! stdio-style tests and lightweight protocol harnesses. Endpoint helpers mutate
3//! in-memory transcripts and codec helpers read or write caller-provided streams.
4//!
5use std::{
6    collections::VecDeque,
7    io::{BufRead, Cursor, Write},
8    sync::{Arc, Mutex},
9};
10
11use agent_sdk_core::AgentError;
12
13use super::JsonRpcId;
14use super::json_rpc::{
15    JsonRpcFrame, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, expect_notification,
16    expect_response, protocol_violation, stdio_error, validate_json_rpc_line,
17};
18
19#[derive(Clone, Debug, Default)]
20/// Protocol json rpc line codec value used by toolkit JSON-RPC adapters.
21/// Constructing the value prepares protocol data; endpoint and transport methods own transcript or I/O effects.
22pub struct JsonRpcLineCodec;
23
24impl JsonRpcLineCodec {
25    /// Writes one newline-delimited JSON-RPC frame or raw line to the
26    /// caller-provided writer. It does not launch a process, open sockets, or
27    /// persist transport state.
28    pub fn write_frame(writer: &mut impl Write, frame: &JsonRpcFrame) -> Result<(), AgentError> {
29        let line = frame.to_line()?;
30        writer.write_all(line.as_bytes()).map_err(stdio_error)?;
31        writer.write_all(b"\n").map_err(stdio_error)?;
32        writer.flush().map_err(stdio_error)
33    }
34
35    /// Reads one newline-delimited JSON-RPC frame or raw line from the
36    /// caller-provided reader. It does not launch a process, open sockets, or
37    /// persist transport state.
38    pub fn read_frame(reader: &mut impl BufRead) -> Result<Option<JsonRpcFrame>, AgentError> {
39        let Some(line) = Self::read_line(reader)? else {
40            return Ok(None);
41        };
42        JsonRpcFrame::from_line(&line).map(Some)
43    }
44
45    /// Writes one newline-delimited JSON-RPC frame or raw line to the
46    /// caller-provided writer. It does not launch a process, open sockets, or
47    /// persist transport state.
48    pub fn write_raw_line(writer: &mut impl Write, line: &str) -> Result<(), AgentError> {
49        validate_json_rpc_line(line)?;
50        writer.write_all(line.as_bytes()).map_err(stdio_error)?;
51        writer.write_all(b"\n").map_err(stdio_error)?;
52        writer.flush().map_err(stdio_error)
53    }
54
55    /// Reads one newline-delimited JSON-RPC frame or raw line from the
56    /// caller-provided reader. It does not launch a process, open sockets, or
57    /// persist transport state.
58    pub fn read_line(reader: &mut impl BufRead) -> Result<Option<String>, AgentError> {
59        let mut line = String::new();
60        let read = reader.read_line(&mut line).map_err(stdio_error)?;
61        if read == 0 {
62            return Ok(None);
63        }
64        if !line.ends_with('\n') {
65            return Err(protocol_violation(
66                "json-rpc stdio frame must be newline-delimited",
67            ));
68        }
69        let line = line.trim_end_matches('\n').trim_end_matches('\r');
70        validate_json_rpc_line(line)?;
71        Ok(Some(line.to_string()))
72    }
73}
74
75#[derive(Clone, Debug)]
76/// Protocol json rpc line endpoint value used by toolkit JSON-RPC adapters.
77/// Constructing the value prepares protocol data; endpoint and transport methods own transcript or I/O effects.
78pub struct JsonRpcLineEndpoint {
79    name: String,
80    incoming: Arc<Mutex<VecDeque<Vec<u8>>>>,
81    outgoing: Arc<Mutex<VecDeque<Vec<u8>>>>,
82    sent_lines: Arc<Mutex<Vec<String>>>,
83    received_lines: Arc<Mutex<Vec<String>>>,
84}
85
86impl JsonRpcLineEndpoint {
87    /// Builds the pair value.
88    /// This is data construction and performs no I/O, journal append, event publication, or
89    /// process work.
90    pub fn pair(left_name: impl Into<String>, right_name: impl Into<String>) -> (Self, Self) {
91        let left_to_right = Arc::new(Mutex::new(VecDeque::new()));
92        let right_to_left = Arc::new(Mutex::new(VecDeque::new()));
93        let left = Self {
94            name: left_name.into(),
95            incoming: right_to_left.clone(),
96            outgoing: left_to_right.clone(),
97            sent_lines: Arc::new(Mutex::new(Vec::new())),
98            received_lines: Arc::new(Mutex::new(Vec::new())),
99        };
100        let right = Self {
101            name: right_name.into(),
102            incoming: left_to_right,
103            outgoing: right_to_left,
104            sent_lines: Arc::new(Mutex::new(Vec::new())),
105            received_lines: Arc::new(Mutex::new(Vec::new())),
106        };
107        (left, right)
108    }
109
110    /// Returns the name currently held by this value.
111    /// This reads endpoint metadata or a queued response from the in-memory transport.
112    pub fn name(&self) -> &str {
113        &self.name
114    }
115
116    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
117    /// transcript state for tests; it performs no OS-level I/O.
118    pub fn send_frame(&self, frame: JsonRpcFrame) -> Result<String, AgentError> {
119        let line = frame.to_line()?;
120        let mut bytes = Vec::new();
121        JsonRpcLineCodec::write_frame(&mut bytes, &frame)?;
122        self.queue_line(line.clone(), bytes)?;
123        Ok(line)
124    }
125
126    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
127    /// transcript state for tests; it performs no OS-level I/O.
128    pub fn send_raw_line(&self, line: impl Into<String>) -> Result<(), AgentError> {
129        let line = line.into();
130        let mut bytes = Vec::new();
131        JsonRpcLineCodec::write_raw_line(&mut bytes, &line)?;
132        self.queue_line(line, bytes)
133    }
134
135    /// Reads a JSON-RPC line or frame from the in-memory endpoint queue. It
136    /// does not perform OS-level I/O.
137    pub fn try_receive_raw_line(&self) -> Result<Option<String>, AgentError> {
138        let Some(bytes) = self
139            .incoming
140            .lock()
141            .map_err(|_| protocol_violation("json-rpc incoming lock poisoned"))?
142            .pop_front()
143        else {
144            return Ok(None);
145        };
146        let mut reader = Cursor::new(bytes);
147        let line = JsonRpcLineCodec::read_line(&mut reader)?;
148        if let Some(line) = &line {
149            self.received_lines
150                .lock()
151                .map_err(|_| protocol_violation("json-rpc received transcript lock poisoned"))?
152                .push(line.clone());
153        }
154        Ok(line)
155    }
156
157    /// Reads a JSON-RPC line or frame from the in-memory endpoint queue. It
158    /// does not perform OS-level I/O.
159    pub fn try_receive_frame(&self) -> Result<Option<JsonRpcFrame>, AgentError> {
160        self.try_receive_raw_line()?
161            .map(|line| JsonRpcFrame::from_line(&line))
162            .transpose()
163    }
164
165    /// Reads a JSON-RPC line or frame from the in-memory endpoint queue. It
166    /// does not perform OS-level I/O.
167    pub fn receive_frame(&self) -> Result<JsonRpcFrame, AgentError> {
168        self.try_receive_frame()?.ok_or_else(|| {
169            protocol_violation(format!("{} has no queued json-rpc frame", self.name))
170        })
171    }
172
173    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
174    /// transcript state for tests; it performs no OS-level I/O.
175    pub fn send_request(
176        &self,
177        id: impl Into<JsonRpcId>,
178        method: impl Into<String>,
179        params: serde_json::Value,
180    ) -> Result<String, AgentError> {
181        self.send_frame(JsonRpcFrame::Request(JsonRpcRequest::new(
182            id, method, params,
183        )))
184    }
185
186    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
187    /// transcript state for tests; it performs no OS-level I/O.
188    pub fn send_notification(
189        &self,
190        method: impl Into<String>,
191        params: serde_json::Value,
192    ) -> Result<String, AgentError> {
193        self.send_frame(JsonRpcFrame::Notification(JsonRpcNotification::new(
194            method, params,
195        )))
196    }
197
198    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
199    /// transcript state for tests; it performs no OS-level I/O.
200    pub fn send_result(
201        &self,
202        id: JsonRpcId,
203        result: serde_json::Value,
204    ) -> Result<String, AgentError> {
205        self.send_frame(JsonRpcFrame::Response(JsonRpcResponse::result(id, result)))
206    }
207
208    /// Queues a JSON-RPC frame on the paired in-memory endpoint and records
209    /// transcript state for tests; it performs no OS-level I/O.
210    pub fn send_error(
211        &self,
212        id: Option<JsonRpcId>,
213        code: i64,
214        message: impl Into<String>,
215    ) -> Result<String, AgentError> {
216        self.send_frame(JsonRpcFrame::Response(JsonRpcResponse::error(
217            id, code, message,
218        )))
219    }
220
221    /// Returns the response currently held by this value.
222    /// This reads endpoint metadata or a queued response from the in-memory transport.
223    pub fn response(&self) -> Result<JsonRpcResponse, AgentError> {
224        expect_response(self.receive_frame()?)
225    }
226
227    /// Returns notification for this protocol::line_transport value without
228    /// performing external I/O.
229    pub fn notification(&self) -> Result<JsonRpcNotification, AgentError> {
230        expect_notification(self.receive_frame()?)
231    }
232
233    /// Returns sent lines for this protocol::line_transport value without
234    /// performing external I/O. Panics only if the in-memory test transcript
235    /// lock is poisoned.
236    pub fn sent_lines(&self) -> Vec<String> {
237        self.sent_lines
238            .lock()
239            .expect("json-rpc sent transcript")
240            .clone()
241    }
242
243    /// Returns received lines for this protocol::line_transport value without
244    /// performing external I/O. Panics only if the in-memory test transcript
245    /// lock is poisoned.
246    pub fn received_lines(&self) -> Vec<String> {
247        self.received_lines
248            .lock()
249            .expect("json-rpc received transcript")
250            .clone()
251    }
252
253    fn queue_line(&self, line: String, bytes: Vec<u8>) -> Result<(), AgentError> {
254        self.sent_lines
255            .lock()
256            .map_err(|_| protocol_violation("json-rpc sent transcript lock poisoned"))?
257            .push(line);
258        self.outgoing
259            .lock()
260            .map_err(|_| protocol_violation("json-rpc outgoing lock poisoned"))?
261            .push_back(bytes);
262        Ok(())
263    }
264}
agent_sdk_toolkit/protocol/line_transport.rs

agent_sdk_toolkit/protocol/
line_transport.rs
