claude_codes/
client_raw_async.rs

1//! Raw asynchronous transport for Claude Code's stream-json protocol.
2//!
3//! This client only frames newline-delimited messages. It does not decode JSON.
4
5use crate::cli::ClaudeCliBuilder;
6use crate::error::{Error, Result};
7use log::{debug, error};
8use serde::Serialize;
9use std::process::ExitStatus;
10use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
11use tokio::process::Child;
12
13const STDOUT_BUFFER_SIZE: usize = 10 * 1024 * 1024;
14
15pub struct RawAsyncClient {
16    child: Child,
17    writer: BufWriter<tokio::process::ChildStdin>,
18    reader: BufReader<tokio::process::ChildStdout>,
19    _stderr_drain: tokio::task::JoinHandle<()>,
20}
21
22impl RawAsyncClient {
23    pub async fn start_with(builder: ClaudeCliBuilder) -> Result<Self> {
24        crate::version::check_claude_version_async().await?;
25        Self::new(builder.spawn().await?)
26    }
27
28    pub fn new(mut child: Child) -> Result<Self> {
29        let stdin = child
30            .stdin
31            .take()
32            .ok_or_else(|| Error::Protocol("Failed to get stdin".to_string()))?;
33        let stdout = child
34            .stdout
35            .take()
36            .ok_or_else(|| Error::Protocol("Failed to get stdout".to_string()))?;
37        let stderr = child
38            .stderr
39            .take()
40            .ok_or_else(|| Error::Protocol("Failed to get stderr".to_string()))?;
41        let stderr_drain = tokio::spawn(async move {
42            let mut stderr = BufReader::new(stderr);
43            let mut line = String::new();
44            loop {
45                line.clear();
46                match stderr.read_line(&mut line).await {
47                    Ok(0) | Err(_) => break,
48                    Ok(_) => debug!("[RAW CLIENT STDERR] {}", line.trim_end()),
49                }
50            }
51        });
52
53        Ok(Self {
54            child,
55            writer: BufWriter::new(stdin),
56            reader: BufReader::with_capacity(STDOUT_BUFFER_SIZE, stdout),
57            _stderr_drain: stderr_drain,
58        })
59    }
60
61    /// Serialize and write one typed SDK message.
62    pub async fn send<T: Serialize>(&mut self, message: &T) -> Result<()> {
63        let line = serde_json::to_string(message).map_err(Error::Json)?;
64        self.write_line(&line).await
65    }
66
67    async fn write_line(&mut self, line: &str) -> Result<()> {
68        let line = single_line(line)?;
69        debug!("[RAW CLIENT] Sending {} bytes", line.len());
70        self.writer
71            .write_all(line.as_bytes())
72            .await
73            .map_err(Error::Io)?;
74        self.writer.write_all(b"\n").await.map_err(Error::Io)?;
75        self.writer.flush().await.map_err(Error::Io)
76    }
77
78    /// Read one complete JSONL frame without inspecting its contents.
79    pub async fn next_line(&mut self) -> Result<Option<String>> {
80        let mut line = String::new();
81        loop {
82            line.clear();
83            if self.reader.read_line(&mut line).await.map_err(Error::Io)? == 0 {
84                return Ok(None);
85            }
86            remove_line_ending(&mut line);
87            if line.trim().is_empty() {
88                continue;
89            }
90            debug!("[RAW CLIENT] Received {} bytes", line.len());
91            return Ok(Some(line));
92        }
93    }
94
95    pub fn pid(&self) -> Option<u32> {
96        self.child.id()
97    }
98
99    pub fn is_alive(&mut self) -> bool {
100        self.child.try_wait().ok().flatten().is_none()
101    }
102
103    pub async fn wait_for_exit(&mut self) -> Result<ExitStatus> {
104        self.child.wait().await.map_err(Error::Io)
105    }
106
107    pub async fn shutdown(mut self) -> Result<()> {
108        self.child.kill().await.map_err(Error::Io)
109    }
110}
111
112impl Drop for RawAsyncClient {
113    fn drop(&mut self) {
114        if self.is_alive() {
115            if let Err(error) = self.child.start_kill() {
116                error!("Failed to kill raw Claude process on drop: {error}");
117            }
118        }
119    }
120}
121
122fn single_line(line: &str) -> Result<&str> {
123    let line = line.trim_end_matches(['\r', '\n']);
124    if line.contains(['\r', '\n']) {
125        return Err(Error::Protocol(
126            "raw Claude frame contains an embedded line break".to_string(),
127        ));
128    }
129    Ok(line)
130}
131
132fn remove_line_ending(line: &mut String) {
133    if line.ends_with('\n') {
134        line.pop();
135        if line.ends_with('\r') {
136            line.pop();
137        }
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    #[test]
146    fn accepts_one_frame_with_a_line_ending() {
147        assert_eq!(
148            single_line("{\"type\":\"user\"}\r\n").unwrap(),
149            "{\"type\":\"user\"}"
150        );
151    }
152
153    #[test]
154    fn rejects_multiple_frames() {
155        assert!(single_line("{\"type\":\"user\"}\n{}").is_err());
156    }
157}
claude_codes/client_raw_async.rs

claude_codes/
client_raw_async.rs
