codex_codes/
client_sync.rs

1//! Synchronous multi-turn client for the Codex app-server.
2//!
3//! Spawns `codex app-server --listen stdio://` and communicates over
4//! newline-delimited JSON-RPC. The connection stays open for multiple
5//! turns until explicitly shut down.
6//!
7//! This is the blocking counterpart to [`crate::client_async::AsyncClient`].
8//! Prefer the async client for applications that already use tokio.
9//!
10//! # Lifecycle
11//!
12//! 1. Create a client with [`SyncClient::start`] (spawns and initializes the app-server)
13//! 2. Call [`SyncClient::thread_start`] to create a conversation session
14//! 3. Call [`SyncClient::turn_start`] to send user input
15//! 4. Iterate over [`SyncClient::events`] until `turn/completed`
16//! 5. Handle approval requests via [`SyncClient::respond`]
17//! 6. Repeat steps 3-5 for follow-up turns
18//!
19//! # Example
20//!
21//! ```ignore
22//! use codex_codes::{SyncClient, ThreadStartParams, TurnStartParams, UserInput, ServerMessage};
23//!
24//! let mut client = SyncClient::start()?;
25//! let thread = client.thread_start(&ThreadStartParams::default())?;
26//!
27//! client.turn_start(&TurnStartParams {
28//!     thread_id: thread.thread_id().to_string(),
29//!     input: vec![UserInput::Text { text: "Hello!".into() }],
30//!     model: None,
31//!     reasoning_effort: None,
32//!     sandbox_policy: None,
33//! })?;
34//!
35//! for result in client.events() {
36//!     match result? {
37//!         ServerMessage::Notification(n) => {
38//!             if let codex_codes::Notification::TurnCompleted(_) = n { break; }
39//!         }
40//!         _ => {}
41//!     }
42//! }
43//! ```
44
45use crate::cli::AppServerBuilder;
46use crate::error::{Error, ParseError, Result};
47use crate::jsonrpc::{
48    JsonRpcError, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, RequestId,
49};
50use crate::messages::{Notification, ServerMessage, ServerRequest};
51use crate::protocol::{
52    ClientInfo, InitializeParams, InitializeResponse, ThreadArchiveParams, ThreadArchiveResponse,
53    ThreadDeleteParams, ThreadDeleteResponse, ThreadForkParams, ThreadForkResponse,
54    ThreadItemsListParams, ThreadItemsListResponse, ThreadResumeParams, ThreadResumeResponse,
55    ThreadRevertParams, ThreadRevertResponse, ThreadStartParams, ThreadStartResponse,
56    ThreadTurnsListParams, ThreadTurnsListResponse, TurnInterruptParams, TurnInterruptResponse,
57    TurnStartParams, TurnStartResponse, TurnSteerParams, TurnSteerResponse,
58};
59use log::{debug, warn};
60use serde::de::DeserializeOwned;
61use serde::Serialize;
62use std::collections::VecDeque;
63use std::io::{BufRead, BufReader, BufWriter, Write};
64use std::process::Child;
65
66/// Buffer size for reading stdout (10MB).
67const STDOUT_BUFFER_SIZE: usize = 10 * 1024 * 1024;
68
69/// Synchronous multi-turn client for the Codex app-server.
70///
71/// Communicates with a long-lived `codex app-server` process via
72/// newline-delimited JSON-RPC over stdio. Manages request/response
73/// correlation and buffers incoming notifications that arrive while
74/// waiting for RPC responses.
75///
76/// The client automatically kills the app-server process when dropped.
77pub struct SyncClient {
78    child: Child,
79    writer: BufWriter<std::process::ChildStdin>,
80    reader: BufReader<std::process::ChildStdout>,
81    /// Handle to the background thread draining the child's stderr pipe.
82    /// Kept alive for the lifetime of the client; the thread exits on EOF
83    /// when the child is killed.
84    _stderr_drain: std::thread::JoinHandle<()>,
85    next_id: i64,
86    buffered: VecDeque<ServerMessage>,
87}
88
89impl SyncClient {
90    /// Create a client from an existing child process.
91    ///
92    /// The child's stdin, stdout, and stderr must all be piped. This does not
93    /// perform the app-server `initialize` handshake or a Codex version check.
94    /// Use [`AppServerBuilder::build_command_sync`] to retain the SDK's
95    /// command-line and stdio configuration while customizing how the process
96    /// is spawned.
97    pub fn new(mut child: Child) -> Result<Self> {
98        let stdin = child
99            .stdin
100            .take()
101            .ok_or_else(|| Error::Protocol("Failed to get stdin".to_string()))?;
102        let stdout = child
103            .stdout
104            .take()
105            .ok_or_else(|| Error::Protocol("Failed to get stdout".to_string()))?;
106        let stderr = child
107            .stderr
108            .take()
109            .ok_or_else(|| Error::Protocol("Failed to get stderr".to_string()))?;
110
111        // The app-server emits ~200 KB/s of tracing to stderr. Without an
112        // active reader, the ~64 KB kernel pipe fills almost instantly and
113        // the child blocks. Drain in the background and route lines through
114        // the `log` crate (see [`crate::stderr_drain`]).
115        let stderr_drain = crate::stderr_drain::spawn_sync(stderr);
116
117        Ok(Self {
118            child,
119            writer: BufWriter::new(stdin),
120            reader: BufReader::with_capacity(STDOUT_BUFFER_SIZE, stdout),
121            _stderr_drain: stderr_drain,
122            next_id: 1,
123            buffered: VecDeque::new(),
124        })
125    }
126
127    /// Start an app-server with default settings.
128    ///
129    /// Spawns `codex app-server --listen stdio://`, performs the required
130    /// `initialize` handshake, and returns a connected client ready for
131    /// `thread_start()`.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error if the `codex` CLI is not installed, the version is
136    /// incompatible, the process fails to start, or the initialization
137    /// handshake fails.
138    pub fn start() -> Result<Self> {
139        Self::start_with(AppServerBuilder::new())
140    }
141
142    /// Start an app-server with a custom [`AppServerBuilder`].
143    ///
144    /// Performs the required `initialize` handshake before returning.
145    /// Use this to configure the binary path, working directory, environment,
146    /// or CLI arguments.
147    ///
148    /// # Errors
149    ///
150    /// Returns an error if the process fails to start, stdio pipes
151    /// cannot be established, or the initialization handshake fails.
152    pub fn start_with(builder: AppServerBuilder) -> Result<Self> {
153        let mut client = Self::spawn(builder)?;
154        client.initialize(&InitializeParams {
155            client_info: ClientInfo {
156                name: "codex-codes".to_string(),
157                version: env!("CARGO_PKG_VERSION").to_string(),
158                title: None,
159            },
160            capabilities: None,
161        })?;
162        Ok(client)
163    }
164
165    /// Spawn an app-server without performing the `initialize` handshake.
166    ///
167    /// Use this if you need to send a custom [`InitializeParams`] (e.g., with
168    /// specific capabilities). You **must** call [`SyncClient::initialize`]
169    /// before any other requests.
170    pub fn spawn(builder: AppServerBuilder) -> Result<Self> {
171        crate::version::check_codex_version()?;
172        Self::new(builder.spawn_sync()?)
173    }
174
175    /// Send a JSON-RPC request and wait for the matching response.
176    ///
177    /// Any notifications or server requests that arrive before the response
178    /// are buffered and can be retrieved via [`SyncClient::next_message`].
179    ///
180    /// # Errors
181    ///
182    /// - [`Error::JsonRpc`] if the server returns a JSON-RPC error
183    /// - [`Error::ServerClosed`] if the connection drops before a response arrives
184    /// - [`Error::Json`] if response deserialization fails
185    pub fn request<P: Serialize, R: DeserializeOwned>(
186        &mut self,
187        method: &str,
188        params: &P,
189    ) -> Result<R> {
190        let id = RequestId::Integer(self.next_id);
191        self.next_id += 1;
192
193        let req = JsonRpcRequest {
194            id: id.clone(),
195            method: method.to_string(),
196            params: Some(serde_json::to_value(params).map_err(Error::Json)?),
197        };
198
199        self.send_raw(&req)?;
200
201        loop {
202            let msg = self.read_message()?;
203            match msg {
204                JsonRpcMessage::Response(resp) if resp.id == id => {
205                    let result: R = serde_json::from_value(resp.result).map_err(Error::Json)?;
206                    return Ok(result);
207                }
208                JsonRpcMessage::Error(err) if err.id == id => {
209                    return Err(Error::JsonRpc {
210                        code: err.error.code,
211                        message: err.error.message,
212                    });
213                }
214                JsonRpcMessage::Notification(notif) => {
215                    let typed = Notification::from_envelope(&notif.method, notif.params)
216                        .map_err(Error::Json)?;
217                    self.buffered.push_back(ServerMessage::Notification(typed));
218                }
219                JsonRpcMessage::Request(req) => {
220                    let typed = ServerRequest::from_envelope(&req.method, req.params)
221                        .map_err(Error::Json)?;
222                    self.buffered.push_back(ServerMessage::Request {
223                        id: req.id,
224                        request: typed,
225                    });
226                }
227                JsonRpcMessage::Response(resp) => {
228                    warn!(
229                        "[CLIENT] Unexpected response for id={}, expected id={}",
230                        resp.id, id
231                    );
232                }
233                JsonRpcMessage::Error(err) => {
234                    warn!(
235                        "[CLIENT] Unexpected error for id={}, expected id={}",
236                        err.id, id
237                    );
238                }
239            }
240        }
241    }
242
243    /// Start a new thread (conversation session).
244    ///
245    /// A thread must be created before any turns can be started. The returned
246    /// [`ThreadStartResponse`] contains the `thread_id` needed for subsequent calls.
247    pub fn thread_start(&mut self, params: &ThreadStartParams) -> Result<ThreadStartResponse> {
248        self.request(crate::protocol::methods::THREAD_START, params)
249    }
250
251    /// Resume a previously persisted thread by id.
252    ///
253    /// Replays the thread's history so turns can continue where they left off.
254    pub fn thread_resume(&mut self, params: &ThreadResumeParams) -> Result<ThreadResumeResponse> {
255        self.request(crate::protocol::methods::THREAD_RESUME, params)
256    }
257
258    /// Fork an existing thread into a new independent thread.
259    pub fn thread_fork(&mut self, params: &ThreadForkParams) -> Result<ThreadForkResponse> {
260        self.request(crate::protocol::methods::THREAD_FORK, params)
261    }
262
263    /// Start a new turn within a thread.
264    ///
265    /// Sends user input to the agent. After calling this, use [`SyncClient::events`]
266    /// or [`SyncClient::next_message`] to consume notifications until `turn/completed`.
267    pub fn turn_start(&mut self, params: &TurnStartParams) -> Result<TurnStartResponse> {
268        self.request(crate::protocol::methods::TURN_START, params)
269    }
270
271    /// Steer an active turn with additional user input (`turn/steer`).
272    pub fn turn_steer(&mut self, params: &TurnSteerParams) -> Result<TurnSteerResponse> {
273        self.request(crate::protocol::methods::TURN_STEER, params)
274    }
275
276    /// Page through a thread's items in canonical order (`thread/items/list`).
277    pub fn thread_items_list(
278        &mut self,
279        params: &ThreadItemsListParams,
280    ) -> Result<ThreadItemsListResponse> {
281        self.request(crate::protocol::methods::THREAD_ITEMS_LIST, params)
282    }
283
284    /// Page through a thread's turns (`thread/turns/list`).
285    pub fn thread_turns_list(
286        &mut self,
287        params: &ThreadTurnsListParams,
288    ) -> Result<ThreadTurnsListResponse> {
289        self.request(crate::protocol::methods::THREAD_TURNS_LIST, params)
290    }
291
292    /// Revert a thread's durable history to before one turn (`thread/revert`).
293    pub fn thread_revert(&mut self, params: &ThreadRevertParams) -> Result<ThreadRevertResponse> {
294        self.request(crate::protocol::methods::THREAD_REVERT, params)
295    }
296
297    /// Interrupt an active turn.
298    pub fn turn_interrupt(
299        &mut self,
300        params: &TurnInterruptParams,
301    ) -> Result<TurnInterruptResponse> {
302        self.request(crate::protocol::methods::TURN_INTERRUPT, params)
303    }
304
305    /// Archive a thread.
306    pub fn thread_archive(
307        &mut self,
308        params: &ThreadArchiveParams,
309    ) -> Result<ThreadArchiveResponse> {
310        self.request(crate::protocol::methods::THREAD_ARCHIVE, params)
311    }
312
313    /// Delete a thread.
314    pub fn thread_delete(&mut self, params: &ThreadDeleteParams) -> Result<ThreadDeleteResponse> {
315        self.request(crate::protocol::methods::THREAD_DELETE, params)
316    }
317
318    /// Perform the `initialize` handshake with the app-server.
319    ///
320    /// Sends `initialize` with the given params and then sends the
321    /// `initialized` notification. This must be the first request after
322    /// spawning the process.
323    pub fn initialize(&mut self, params: &InitializeParams) -> Result<InitializeResponse> {
324        let resp: InitializeResponse =
325            self.request(crate::protocol::methods::INITIALIZE, params)?;
326        self.send_notification(crate::protocol::methods::INITIALIZED)?;
327        Ok(resp)
328    }
329
330    /// Respond to a server-to-client request (e.g., approval flow).
331    ///
332    /// When the server sends a [`ServerMessage::Request`], it expects a response.
333    /// Use this method with the request's `id` and a result payload. For command
334    /// approval, pass a [`CommandExecutionApprovalResponse`](crate::CommandExecutionApprovalResponse).
335    /// For file change approval, pass a [`FileChangeApprovalResponse`](crate::FileChangeApprovalResponse).
336    pub fn respond<R: Serialize>(&mut self, id: RequestId, result: &R) -> Result<()> {
337        let resp = JsonRpcResponse {
338            id,
339            result: serde_json::to_value(result).map_err(Error::Json)?,
340        };
341        self.send_raw(&resp)
342    }
343
344    /// Respond to a server-to-client request with an error.
345    pub fn respond_error(&mut self, id: RequestId, code: i64, message: &str) -> Result<()> {
346        let err = JsonRpcError {
347            id,
348            error: crate::jsonrpc::JsonRpcErrorData {
349                code,
350                message: message.to_string(),
351                data: None,
352            },
353        };
354        self.send_raw(&err)
355    }
356
357    /// Read the next incoming server message (notification or server request).
358    ///
359    /// Returns buffered messages first (from notifications that arrived during
360    /// a [`SyncClient::request`] call), then reads from the wire.
361    ///
362    /// Returns `Ok(None)` when the app-server closes the connection (EOF).
363    pub fn next_message(&mut self) -> Result<Option<ServerMessage>> {
364        if let Some(msg) = self.buffered.pop_front() {
365            return Ok(Some(msg));
366        }
367
368        loop {
369            let msg = match self.read_message_opt()? {
370                Some(m) => m,
371                None => return Ok(None),
372            };
373
374            match msg {
375                JsonRpcMessage::Notification(notif) => {
376                    let JsonRpcNotification { method, params } = notif;
377                    let typed =
378                        Notification::from_envelope(&method, params.clone()).map_err(|e| {
379                            Error::Deserialization(ParseError::from_envelope(method, params, e))
380                        })?;
381                    return Ok(Some(ServerMessage::Notification(typed)));
382                }
383                JsonRpcMessage::Request(req) => {
384                    let JsonRpcRequest { id, method, params } = req;
385                    let typed =
386                        ServerRequest::from_envelope(&method, params.clone()).map_err(|e| {
387                            Error::Deserialization(ParseError::from_envelope(method, params, e))
388                        })?;
389                    return Ok(Some(ServerMessage::Request { id, request: typed }));
390                }
391                JsonRpcMessage::Response(resp) => {
392                    warn!(
393                        "[CLIENT] Unexpected response (no pending request): id={}",
394                        resp.id
395                    );
396                }
397                JsonRpcMessage::Error(err) => {
398                    warn!(
399                        "[CLIENT] Unexpected error (no pending request): id={} code={}",
400                        err.id, err.error.code
401                    );
402                }
403            }
404        }
405    }
406
407    /// Return an iterator over [`ServerMessage`]s.
408    ///
409    /// The iterator yields `Result<ServerMessage>` and terminates when the
410    /// connection closes (EOF). This is the idiomatic way to consume a turn's
411    /// notifications in synchronous code.
412    pub fn events(&mut self) -> EventIterator<'_> {
413        EventIterator { client: self }
414    }
415
416    /// Shut down the child process.
417    ///
418    /// Kills the process if it's still running. Called automatically on [`Drop`].
419    pub fn shutdown(&mut self) -> Result<()> {
420        debug!("[CLIENT] Shutting down");
421        match self.child.try_wait() {
422            Ok(Some(_)) => Ok(()),
423            Ok(None) => {
424                self.child.kill().map_err(Error::Io)?;
425                self.child.wait().map_err(Error::Io)?;
426                Ok(())
427            }
428            Err(e) => Err(Error::Io(e)),
429        }
430    }
431
432    // -- internal --
433
434    fn send_notification(&mut self, method: &str) -> Result<()> {
435        let notif = JsonRpcNotification {
436            method: method.to_string(),
437            params: None,
438        };
439        self.send_raw(&notif)
440    }
441
442    fn send_raw<T: Serialize>(&mut self, msg: &T) -> Result<()> {
443        let json = serde_json::to_string(msg).map_err(Error::Json)?;
444        debug!("[CLIENT] Sending: {}", json);
445        self.writer.write_all(json.as_bytes()).map_err(Error::Io)?;
446        self.writer.write_all(b"\n").map_err(Error::Io)?;
447        self.writer.flush().map_err(Error::Io)?;
448        Ok(())
449    }
450
451    fn read_message(&mut self) -> Result<JsonRpcMessage> {
452        self.read_message_opt()?.ok_or(Error::ServerClosed)
453    }
454
455    fn read_message_opt(&mut self) -> Result<Option<JsonRpcMessage>> {
456        loop {
457            let mut line = String::new();
458            match self.reader.read_line(&mut line) {
459                Ok(0) => {
460                    debug!("[CLIENT] Stream closed (EOF)");
461                    return Ok(None);
462                }
463                Ok(_) => {
464                    let trimmed = line.trim();
465                    if trimmed.is_empty() {
466                        continue;
467                    }
468
469                    debug!("[CLIENT] Received: {}", trimmed);
470
471                    match serde_json::from_str::<JsonRpcMessage>(trimmed) {
472                        Ok(msg) => return Ok(Some(msg)),
473                        Err(e) => {
474                            warn!(
475                                "[CLIENT] Failed to deserialize message. \
476                                 Please report this at https://github.com/meawoppl/rust-code-agent-sdks/issues"
477                            );
478                            warn!("[CLIENT] Parse error: {}", e);
479                            warn!("[CLIENT] Raw: {}", trimmed);
480                            return Err(Error::Deserialization(ParseError::from_line(trimmed, e)));
481                        }
482                    }
483                }
484                Err(e) => {
485                    debug!("[CLIENT] Error reading stdout: {}", e);
486                    return Err(Error::Io(e));
487                }
488            }
489        }
490    }
491}
492
493impl Drop for SyncClient {
494    fn drop(&mut self) {
495        if let Err(e) = self.shutdown() {
496            debug!("[CLIENT] Error during shutdown: {}", e);
497        }
498    }
499}
500
501/// Iterator over [`ServerMessage`]s from a [`SyncClient`].
502pub struct EventIterator<'a> {
503    client: &'a mut SyncClient,
504}
505
506impl Iterator for EventIterator<'_> {
507    type Item = Result<ServerMessage>;
508
509    fn next(&mut self) -> Option<Self::Item> {
510        match self.client.next_message() {
511            Ok(Some(msg)) => Some(Ok(msg)),
512            Ok(None) => None,
513            Err(e) => Some(Err(e)),
514        }
515    }
516}
517
518#[cfg(test)]
519mod tests {
520    use super::*;
521
522    #[test]
523    fn test_buffer_size() {
524        assert_eq!(STDOUT_BUFFER_SIZE, 10 * 1024 * 1024);
525    }
526}
codex_codes/client_sync.rs

codex_codes/
client_sync.rs
