codex_codes/
client_async.rs

1//! Asynchronous 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//! # Lifecycle
8//!
9//! 1. Create a client with [`AsyncClient::start`] (spawns and initializes the app-server)
10//! 2. Call [`AsyncClient::thread_start`] to create a conversation session
11//! 3. Call [`AsyncClient::turn_start`] to send user input
12//! 4. Consume [`AsyncClient::next_message`] to stream notifications
13//! 5. Handle approval requests via [`AsyncClient::respond`]
14//! 6. Repeat steps 3-5 for follow-up turns
15//! 7. The client kills the app-server on [`Drop`]
16//!
17//! # Example
18//!
19//! ```ignore
20//! use codex_codes::{AsyncClient, ThreadStartParams, TurnStartParams, UserInput, ServerMessage};
21//!
22//! let mut client = AsyncClient::start().await?;
23//! let thread = client.thread_start(&ThreadStartParams::default()).await?;
24//!
25//! client.turn_start(&TurnStartParams {
26//!     thread_id: thread.thread_id().to_string(),
27//!     input: vec![UserInput::Text { text: "Hello!".into() }],
28//!     model: None,
29//!     reasoning_effort: None,
30//!     sandbox_policy: None,
31//! }).await?;
32//!
33//! while let Some(msg) = client.next_message().await? {
34//!     match msg {
35//!         ServerMessage::Notification(n) => {
36//!             if let codex_codes::Notification::TurnCompleted(_) = n { break; }
37//!         }
38//!         ServerMessage::Request { id, .. } => {
39//!             client.respond(id, &serde_json::json!({"decision": "accept"})).await?;
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    ThreadResumeParams, ThreadResumeResponse, ThreadStartParams, ThreadStartResponse,
55    TurnInterruptParams, TurnInterruptResponse, TurnStartParams, TurnStartResponse,
56};
57use log::{debug, error, warn};
58use serde::de::DeserializeOwned;
59use serde::Serialize;
60use std::collections::VecDeque;
61use std::sync::atomic::{AtomicI64, Ordering};
62use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
63use tokio::process::Child;
64
65/// Buffer size for reading stdout (10MB).
66const STDOUT_BUFFER_SIZE: usize = 10 * 1024 * 1024;
67
68/// Asynchronous multi-turn client for the Codex app-server.
69///
70/// Communicates with a long-lived `codex app-server` process via
71/// newline-delimited JSON-RPC over stdio. Manages request/response
72/// correlation and buffers incoming notifications that arrive while
73/// waiting for RPC responses.
74///
75/// The client automatically kills the app-server process when dropped.
76pub struct AsyncClient {
77    child: Child,
78    writer: BufWriter<tokio::process::ChildStdin>,
79    reader: BufReader<tokio::process::ChildStdout>,
80    /// Handle to the background task draining the child's stderr pipe.
81    /// Kept alive for the lifetime of the client; the task exits on EOF
82    /// when the child is killed.
83    _stderr_drain: tokio::task::JoinHandle<()>,
84    next_id: AtomicI64,
85    /// Buffered incoming messages (notifications/server requests) that arrived
86    /// while waiting for a response to a client request.
87    buffered: VecDeque<ServerMessage>,
88}
89
90impl AsyncClient {
91    /// Create a client from an existing Tokio child process.
92    ///
93    /// The child's stdin, stdout, and stderr must all be piped. This does not
94    /// perform the app-server `initialize` handshake or a Codex version check.
95    /// Use [`AppServerBuilder::build_command`] to retain the SDK's command-line
96    /// and stdio configuration while customizing how the process 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_async(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: AtomicI64::new(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 async fn start() -> Result<Self> {
139        Self::start_with(AppServerBuilder::new()).await
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 async fn start_with(builder: AppServerBuilder) -> Result<Self> {
153        let mut client = Self::spawn(builder).await?;
154        client
155            .initialize(&InitializeParams {
156                client_info: ClientInfo {
157                    name: "codex-codes".to_string(),
158                    version: env!("CARGO_PKG_VERSION").to_string(),
159                    title: None,
160                },
161                capabilities: None,
162            })
163            .await?;
164        Ok(client)
165    }
166
167    /// Spawn an app-server without performing the `initialize` handshake.
168    ///
169    /// Use this if you need to send a custom [`InitializeParams`] (e.g., with
170    /// specific capabilities). You **must** call [`AsyncClient::initialize`]
171    /// before any other requests.
172    pub async fn spawn(builder: AppServerBuilder) -> Result<Self> {
173        crate::version::check_codex_version_async().await?;
174        Self::new(builder.spawn().await?)
175    }
176
177    /// Send a JSON-RPC request and wait for the matching response.
178    ///
179    /// Any notifications or server requests that arrive before the response
180    /// are buffered and can be retrieved via [`AsyncClient::next_message`].
181    ///
182    /// # Errors
183    ///
184    /// - [`Error::JsonRpc`] if the server returns a JSON-RPC error
185    /// - [`Error::ServerClosed`] if the connection drops before a response arrives
186    /// - [`Error::Json`] if response deserialization fails
187    pub async fn request<P: Serialize, R: DeserializeOwned>(
188        &mut self,
189        method: &str,
190        params: &P,
191    ) -> Result<R> {
192        let id = RequestId::Integer(self.next_id.fetch_add(1, Ordering::Relaxed));
193
194        let req = JsonRpcRequest {
195            id: id.clone(),
196            method: method.to_string(),
197            params: Some(serde_json::to_value(params).map_err(Error::Json)?),
198        };
199
200        self.send_raw(&req).await?;
201
202        // Read lines until we get a response matching our id
203        loop {
204            let msg = self.read_message().await?;
205            match msg {
206                JsonRpcMessage::Response(resp) if resp.id == id => {
207                    let result: R = serde_json::from_value(resp.result).map_err(Error::Json)?;
208                    return Ok(result);
209                }
210                JsonRpcMessage::Error(err) if err.id == id => {
211                    return Err(Error::JsonRpc {
212                        code: err.error.code,
213                        message: err.error.message,
214                    });
215                }
216                // Buffer notifications and server requests
217                JsonRpcMessage::Notification(notif) => {
218                    let typed = Notification::from_envelope(&notif.method, notif.params)
219                        .map_err(Error::Json)?;
220                    self.buffered.push_back(ServerMessage::Notification(typed));
221                }
222                JsonRpcMessage::Request(req) => {
223                    let typed = ServerRequest::from_envelope(&req.method, req.params)
224                        .map_err(Error::Json)?;
225                    self.buffered.push_back(ServerMessage::Request {
226                        id: req.id,
227                        request: typed,
228                    });
229                }
230                // Response/error for a different id — unexpected
231                JsonRpcMessage::Response(resp) => {
232                    warn!(
233                        "[CLIENT] Unexpected response for id={}, expected id={}",
234                        resp.id, id
235                    );
236                }
237                JsonRpcMessage::Error(err) => {
238                    warn!(
239                        "[CLIENT] Unexpected error for id={}, expected id={}",
240                        err.id, id
241                    );
242                }
243            }
244        }
245    }
246
247    /// Start a new thread (conversation session).
248    ///
249    /// A thread must be created before any turns can be started. The returned
250    /// [`ThreadStartResponse`] contains the `thread_id` needed for subsequent calls.
251    pub async fn thread_start(
252        &mut self,
253        params: &ThreadStartParams,
254    ) -> Result<ThreadStartResponse> {
255        self.request(crate::protocol::methods::THREAD_START, params)
256            .await
257    }
258
259    /// Resume a previously persisted thread by id.
260    ///
261    /// Replays the thread's history so turns can continue where they left off.
262    pub async fn thread_resume(
263        &mut self,
264        params: &ThreadResumeParams,
265    ) -> Result<ThreadResumeResponse> {
266        self.request(crate::protocol::methods::THREAD_RESUME, params)
267            .await
268    }
269
270    /// Fork an existing thread into a new independent thread.
271    pub async fn thread_fork(&mut self, params: &ThreadForkParams) -> Result<ThreadForkResponse> {
272        self.request(crate::protocol::methods::THREAD_FORK, params)
273            .await
274    }
275
276    /// Start a new turn within a thread.
277    ///
278    /// Sends user input to the agent. After calling this, use [`AsyncClient::next_message`]
279    /// to stream notifications until `turn/completed` arrives.
280    pub async fn turn_start(&mut self, params: &TurnStartParams) -> Result<TurnStartResponse> {
281        self.request(crate::protocol::methods::TURN_START, params)
282            .await
283    }
284
285    /// Interrupt an active turn.
286    pub async fn turn_interrupt(
287        &mut self,
288        params: &TurnInterruptParams,
289    ) -> Result<TurnInterruptResponse> {
290        self.request(crate::protocol::methods::TURN_INTERRUPT, params)
291            .await
292    }
293
294    /// Archive a thread.
295    pub async fn thread_archive(
296        &mut self,
297        params: &ThreadArchiveParams,
298    ) -> Result<ThreadArchiveResponse> {
299        self.request(crate::protocol::methods::THREAD_ARCHIVE, params)
300            .await
301    }
302
303    /// Delete a thread.
304    pub async fn thread_delete(
305        &mut self,
306        params: &ThreadDeleteParams,
307    ) -> Result<ThreadDeleteResponse> {
308        self.request(crate::protocol::methods::THREAD_DELETE, params)
309            .await
310    }
311
312    /// Perform the `initialize` handshake with the app-server.
313    ///
314    /// Sends `initialize` with the given params and then sends the
315    /// `initialized` notification. This must be the first request after
316    /// spawning the process.
317    pub async fn initialize(&mut self, params: &InitializeParams) -> Result<InitializeResponse> {
318        let resp: InitializeResponse = self
319            .request(crate::protocol::methods::INITIALIZE, params)
320            .await?;
321        self.send_notification(crate::protocol::methods::INITIALIZED)
322            .await?;
323        Ok(resp)
324    }
325
326    /// Respond to a server-to-client request (e.g., approval flow).
327    ///
328    /// When the server sends a [`ServerMessage::Request`], it expects a response.
329    /// Use this method with the request's `id` and a result payload. For command
330    /// approval, pass a [`CommandExecutionApprovalResponse`](crate::CommandExecutionApprovalResponse).
331    /// For file change approval, pass a [`FileChangeApprovalResponse`](crate::FileChangeApprovalResponse).
332    pub async fn respond<R: Serialize>(&mut self, id: RequestId, result: &R) -> Result<()> {
333        let resp = JsonRpcResponse {
334            id,
335            result: serde_json::to_value(result).map_err(Error::Json)?,
336        };
337        self.send_raw(&resp).await
338    }
339
340    /// Respond to a server-to-client request with an error.
341    pub async fn respond_error(&mut self, id: RequestId, code: i64, message: &str) -> Result<()> {
342        let err = JsonRpcError {
343            id,
344            error: crate::jsonrpc::JsonRpcErrorData {
345                code,
346                message: message.to_string(),
347                data: None,
348            },
349        };
350        self.send_raw(&err).await
351    }
352
353    /// Read the next incoming server message (notification or server request).
354    ///
355    /// Returns buffered messages first (from notifications that arrived during
356    /// an [`AsyncClient::request`] call), then reads from the wire.
357    ///
358    /// Returns `Ok(None)` when the app-server closes the connection (EOF).
359    ///
360    /// # Typical notification methods
361    ///
362    /// | Method | Meaning |
363    /// |--------|---------|
364    /// | `turn/started` | Agent began processing |
365    /// | `item/agentMessage/delta` | Streaming text chunk |
366    /// | `item/commandExecution/outputDelta` | Command output chunk |
367    /// | `item/started` / `item/completed` | Item lifecycle |
368    /// | `turn/completed` | Agent finished the turn |
369    /// | `error` | Server-side error |
370    pub async fn next_message(&mut self) -> Result<Option<ServerMessage>> {
371        // Drain buffered messages first
372        if let Some(msg) = self.buffered.pop_front() {
373            return Ok(Some(msg));
374        }
375
376        // Read from the wire
377        loop {
378            let msg = match self.read_message_opt().await? {
379                Some(m) => m,
380                None => return Ok(None),
381            };
382
383            match msg {
384                JsonRpcMessage::Notification(notif) => {
385                    let JsonRpcNotification { method, params } = notif;
386                    let typed =
387                        Notification::from_envelope(&method, params.clone()).map_err(|e| {
388                            Error::Deserialization(ParseError::from_envelope(method, params, e))
389                        })?;
390                    return Ok(Some(ServerMessage::Notification(typed)));
391                }
392                JsonRpcMessage::Request(req) => {
393                    let JsonRpcRequest { id, method, params } = req;
394                    let typed =
395                        ServerRequest::from_envelope(&method, params.clone()).map_err(|e| {
396                            Error::Deserialization(ParseError::from_envelope(method, params, e))
397                        })?;
398                    return Ok(Some(ServerMessage::Request { id, request: typed }));
399                }
400                // Unexpected responses without a pending request
401                JsonRpcMessage::Response(resp) => {
402                    warn!(
403                        "[CLIENT] Unexpected response (no pending request): id={}",
404                        resp.id
405                    );
406                }
407                JsonRpcMessage::Error(err) => {
408                    warn!(
409                        "[CLIENT] Unexpected error (no pending request): id={} code={}",
410                        err.id, err.error.code
411                    );
412                }
413            }
414        }
415    }
416
417    /// Return an async event stream over [`ServerMessage`]s.
418    ///
419    /// Wraps [`AsyncClient::next_message`] in a stream-like API. Call
420    /// [`EventStream::next`] in a loop, or [`EventStream::collect`] to
421    /// gather all messages until EOF.
422    pub fn events(&mut self) -> EventStream<'_> {
423        EventStream { client: self }
424    }
425
426    /// Get the process ID.
427    pub fn pid(&self) -> Option<u32> {
428        self.child.id()
429    }
430
431    /// Check if the child process is still running.
432    pub fn is_alive(&mut self) -> bool {
433        self.child.try_wait().ok().flatten().is_none()
434    }
435
436    /// Shut down the app-server process.
437    ///
438    /// Consumes the client. If you don't call this explicitly, the
439    /// [`Drop`] implementation will kill the process automatically.
440    pub async fn shutdown(mut self) -> Result<()> {
441        debug!("[CLIENT] Shutting down");
442        self.child.kill().await.map_err(Error::Io)?;
443        Ok(())
444    }
445
446    // -- internal --
447
448    async fn send_notification(&mut self, method: &str) -> Result<()> {
449        let notif = JsonRpcNotification {
450            method: method.to_string(),
451            params: None,
452        };
453        self.send_raw(&notif).await
454    }
455
456    async fn send_raw<T: Serialize>(&mut self, msg: &T) -> Result<()> {
457        let json = serde_json::to_string(msg).map_err(Error::Json)?;
458        debug!("[CLIENT] Sending: {}", json);
459        self.writer
460            .write_all(json.as_bytes())
461            .await
462            .map_err(Error::Io)?;
463        self.writer.write_all(b"\n").await.map_err(Error::Io)?;
464        self.writer.flush().await.map_err(Error::Io)?;
465        Ok(())
466    }
467
468    async fn read_message(&mut self) -> Result<JsonRpcMessage> {
469        self.read_message_opt().await?.ok_or(Error::ServerClosed)
470    }
471
472    async fn read_message_opt(&mut self) -> Result<Option<JsonRpcMessage>> {
473        let mut line = String::new();
474
475        loop {
476            line.clear();
477            let bytes_read = self.reader.read_line(&mut line).await.map_err(Error::Io)?;
478
479            if bytes_read == 0 {
480                debug!("[CLIENT] Stream closed (EOF)");
481                return Ok(None);
482            }
483
484            let trimmed = line.trim();
485            if trimmed.is_empty() {
486                continue;
487            }
488
489            debug!("[CLIENT] Received: {}", trimmed);
490
491            match serde_json::from_str::<JsonRpcMessage>(trimmed) {
492                Ok(msg) => return Ok(Some(msg)),
493                Err(e) => {
494                    warn!(
495                        "[CLIENT] Failed to deserialize message. \
496                         Please report this at https://github.com/meawoppl/rust-code-agent-sdks/issues"
497                    );
498                    warn!("[CLIENT] Parse error: {}", e);
499                    warn!("[CLIENT] Raw: {}", trimmed);
500                    return Err(Error::Deserialization(ParseError::from_line(trimmed, e)));
501                }
502            }
503        }
504    }
505}
506
507impl Drop for AsyncClient {
508    fn drop(&mut self) {
509        if self.is_alive() {
510            if let Err(e) = self.child.start_kill() {
511                error!("Failed to kill app-server process on drop: {}", e);
512            }
513        }
514    }
515}
516
517/// Async stream of [`ServerMessage`]s from an [`AsyncClient`].
518pub struct EventStream<'a> {
519    client: &'a mut AsyncClient,
520}
521
522impl EventStream<'_> {
523    /// Get the next server message.
524    pub async fn next(&mut self) -> Option<Result<ServerMessage>> {
525        match self.client.next_message().await {
526            Ok(Some(msg)) => Some(Ok(msg)),
527            Ok(None) => None,
528            Err(e) => Some(Err(e)),
529        }
530    }
531
532    /// Collect all remaining messages.
533    pub async fn collect(mut self) -> Result<Vec<ServerMessage>> {
534        let mut msgs = Vec::new();
535        while let Some(result) = self.next().await {
536            msgs.push(result?);
537        }
538        Ok(msgs)
539    }
540}
541
542#[cfg(test)]
543mod tests {
544    use super::*;
545
546    #[test]
547    fn test_buffer_size() {
548        assert_eq!(STDOUT_BUFFER_SIZE, 10 * 1024 * 1024);
549    }
550}
codex_codes/client_async.rs

codex_codes/
client_async.rs
