cccc_sdk/
client.rs

1use std::collections::BTreeMap;
2use std::io::{BufRead, BufReader, Read, Write};
3use std::net::{TcpStream, ToSocketAddrs};
4use std::path::Path;
5use std::time::Duration;
6
7use serde::de::DeserializeOwned;
8use serde_json::{json, Map, Value};
9
10use crate::{
11    discover_endpoint, DaemonEndpoint, DaemonRequest, DaemonResponse, Error, PingResult, Result,
12};
13
14const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
15const MAX_REQUEST_BYTES: usize = 2_000_000;
16const MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
17
18/// Capabilities and operations required by [`CCCCClient::assert_compatible`].
19#[derive(Clone, Debug, Default)]
20pub struct CompatibilityRequirements<'a> {
21    pub minimum_ipc_version: u32,
22    pub capabilities: BTreeMap<&'a str, bool>,
23    pub operations: Vec<&'a str>,
24}
25
26/// Blocking CCCC Daemon IPC v1 client.
27#[derive(Clone, Debug)]
28pub struct CCCCClient {
29    endpoint: DaemonEndpoint,
30    timeout: Duration,
31}
32
33impl CCCCClient {
34    /// Discover the currently running daemon.
35    pub fn discover() -> Result<Self> {
36        Self::discover_in(None)
37    }
38
39    /// Discover a daemon under an explicit `CCCC_HOME`.
40    pub fn discover_in(cccc_home: Option<&Path>) -> Result<Self> {
41        Ok(Self::new(discover_endpoint(cccc_home)?))
42    }
43
44    pub fn new(endpoint: DaemonEndpoint) -> Self {
45        Self {
46            endpoint,
47            timeout: DEFAULT_TIMEOUT,
48        }
49    }
50
51    pub fn with_timeout(mut self, timeout: Duration) -> Self {
52        self.timeout = timeout;
53        self
54    }
55
56    pub fn endpoint(&self) -> &DaemonEndpoint {
57        &self.endpoint
58    }
59
60    /// Call any non-streaming daemon operation and return its result object.
61    pub fn call(&self, op: &str, args: Map<String, Value>) -> Result<Map<String, Value>> {
62        let response = self.call_raw(op, args)?;
63        if response.v != 1 {
64            return Err(Error::UnsupportedIpcVersion(response.v));
65        }
66        if response.ok {
67            return Ok(response.result);
68        }
69        Err(Error::Daemon(response.error.unwrap_or_else(|| {
70            crate::DaemonError {
71                code: "error".into(),
72                message: "daemon returned ok=false without an error".into(),
73                details: json!({}),
74            }
75        })))
76    }
77
78    /// Call an operation and retain the full response envelope.
79    pub fn call_raw(&self, op: &str, args: Map<String, Value>) -> Result<DaemonResponse> {
80        if op.trim().is_empty() {
81            return Err(Error::Incompatible("operation name cannot be empty".into()));
82        }
83        let request = DaemonRequest {
84            v: 1,
85            op,
86            args: &args,
87        };
88        let mut encoded = serde_json::to_vec(&request)?;
89        encoded.push(b'\n');
90        if encoded.len() > MAX_REQUEST_BYTES {
91            return Err(Error::RequestTooLarge(MAX_REQUEST_BYTES));
92        }
93
94        match &self.endpoint {
95            DaemonEndpoint::Tcp { host, port } => {
96                let address = (host.as_str(), *port)
97                    .to_socket_addrs()?
98                    .next()
99                    .ok_or_else(|| {
100                        Error::InvalidEndpoint(format!("cannot resolve {host}:{port}"))
101                    })?;
102                let mut stream = TcpStream::connect_timeout(&address, self.timeout)?;
103                stream.set_read_timeout(Some(self.timeout))?;
104                stream.set_write_timeout(Some(self.timeout))?;
105                exchange(&mut stream, &encoded)
106            }
107            DaemonEndpoint::Unix(path) => self.call_unix(path, &encoded),
108        }
109    }
110
111    #[cfg(unix)]
112    fn call_unix(&self, path: &Path, encoded: &[u8]) -> Result<DaemonResponse> {
113        use std::os::unix::net::UnixStream;
114
115        let mut stream = UnixStream::connect(path)?;
116        stream.set_read_timeout(Some(self.timeout))?;
117        stream.set_write_timeout(Some(self.timeout))?;
118        exchange(&mut stream, encoded)
119    }
120
121    #[cfg(not(unix))]
122    fn call_unix(&self, _path: &Path, _encoded: &[u8]) -> Result<DaemonResponse> {
123        Err(Error::UnixSocketUnsupported)
124    }
125
126    pub fn ping(&self) -> Result<PingResult> {
127        self.call_typed("ping", Map::new())
128    }
129
130    pub fn groups(&self) -> Result<Map<String, Value>> {
131        self.call("groups", Map::new())
132    }
133
134    pub fn group_show(&self, group_id: &str) -> Result<Map<String, Value>> {
135        self.call("group_show", object([("group_id", json!(group_id))]))
136    }
137
138    pub fn send(&self, group_id: &str, text: &str, by: &str) -> Result<Map<String, Value>> {
139        self.call(
140            "send",
141            object([
142                ("group_id", json!(group_id)),
143                ("text", json!(text)),
144                ("by", json!(by)),
145            ]),
146        )
147    }
148
149    pub fn reply(
150        &self,
151        group_id: &str,
152        reply_to: &str,
153        text: &str,
154        by: &str,
155    ) -> Result<Map<String, Value>> {
156        self.call(
157            "reply",
158            object([
159                ("group_id", json!(group_id)),
160                ("reply_to", json!(reply_to)),
161                ("text", json!(text)),
162                ("by", json!(by)),
163            ]),
164        )
165    }
166
167    pub fn inbox_list(
168        &self,
169        group_id: &str,
170        actor_id: &str,
171        limit: Option<u32>,
172    ) -> Result<Map<String, Value>> {
173        let mut args = object([("group_id", json!(group_id)), ("actor_id", json!(actor_id))]);
174        if let Some(limit) = limit {
175            args.insert("limit".into(), json!(limit));
176        }
177        self.call("inbox_list", args)
178    }
179
180    pub fn context_get(&self, group_id: &str) -> Result<Map<String, Value>> {
181        self.call("context_get", object([("group_id", json!(group_id))]))
182    }
183
184    pub fn context_sync(
185        &self,
186        group_id: &str,
187        by: &str,
188        operations: Vec<Value>,
189    ) -> Result<Map<String, Value>> {
190        self.call(
191            "context_sync",
192            object([
193                ("group_id", json!(group_id)),
194                ("by", json!(by)),
195                ("ops", Value::Array(operations)),
196            ]),
197        )
198    }
199
200    /// Validate protocol, advertised capabilities, and actual op recognition.
201    pub fn assert_compatible(
202        &self,
203        requirements: &CompatibilityRequirements<'_>,
204    ) -> Result<PingResult> {
205        let ping = self.ping()?;
206        let minimum = requirements.minimum_ipc_version.max(1);
207        if ping.ipc_v < minimum {
208            return Err(Error::Incompatible(format!(
209                "daemon ipc_v={} is below required ipc_v={minimum}",
210                ping.ipc_v
211            )));
212        }
213        for (name, required) in &requirements.capabilities {
214            if *required && ping.capabilities.get(*name).and_then(Value::as_bool) != Some(true) {
215                return Err(Error::Incompatible(format!(
216                    "daemon capability {name}=true is required"
217                )));
218            }
219        }
220        for operation in &requirements.operations {
221            if operation_probe_is_unsafe(operation) {
222                continue;
223            }
224            match self.call(operation, Map::new()) {
225                Err(Error::Daemon(error)) if error.code == "unknown_op" => {
226                    return Err(Error::Incompatible(format!(
227                        "daemon does not support operation {operation}"
228                    )));
229                }
230                Err(Error::Daemon(_)) | Ok(_) => {}
231                Err(error) => return Err(error),
232            }
233        }
234        Ok(ping)
235    }
236
237    fn call_typed<T: DeserializeOwned>(&self, op: &str, args: Map<String, Value>) -> Result<T> {
238        let result = self.call(op, args)?;
239        Ok(serde_json::from_value(Value::Object(result))?)
240    }
241}
242
243fn operation_probe_is_unsafe(operation: &str) -> bool {
244    matches!(
245        operation,
246        "ping"
247            | "shutdown"
248            | "term_attach"
249            | "presentation_browser_attach"
250            | "presentation_browser_vnc_attach"
251            | "web_model_browser_attach"
252            | "web_model_browser_vnc_attach"
253            | "space_provider_auth_browser_attach"
254            | "space_provider_auth_browser_vnc_attach"
255            | "runtime_hermes_prepare"
256            | "runtime_hermes_mcp_test"
257    )
258}
259
260fn object<const N: usize>(entries: [(&str, Value); N]) -> Map<String, Value> {
261    entries
262        .into_iter()
263        .map(|(key, value)| (key.to_owned(), value))
264        .collect()
265}
266
267fn exchange<S: Read + Write>(stream: &mut S, request: &[u8]) -> Result<DaemonResponse> {
268    stream.write_all(request)?;
269    stream.flush()?;
270
271    let mut reader = BufReader::new(stream);
272    let mut response = Vec::new();
273    let read = reader
274        .by_ref()
275        .take((MAX_RESPONSE_BYTES + 1) as u64)
276        .read_until(b'\n', &mut response)?;
277    if read == 0 {
278        return Err(Error::EmptyResponse);
279    }
280    if response.len() > MAX_RESPONSE_BYTES {
281        return Err(Error::ResponseTooLarge(MAX_RESPONSE_BYTES));
282    }
283    Ok(serde_json::from_slice(&response)?)
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289    use std::net::TcpListener;
290    use std::thread;
291
292    fn server_once(response: &'static str) -> (DaemonEndpoint, thread::JoinHandle<String>) {
293        let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server");
294        let address = listener.local_addr().expect("local address");
295        let handle = thread::spawn(move || {
296            let (mut stream, _) = listener.accept().expect("accept client");
297            let mut request = String::new();
298            BufReader::new(&mut stream)
299                .read_line(&mut request)
300                .expect("read request");
301            stream
302                .write_all(response.as_bytes())
303                .expect("write response");
304            request
305        });
306        (
307            DaemonEndpoint::Tcp {
308                host: "127.0.0.1".into(),
309                port: address.port(),
310            },
311            handle,
312        )
313    }
314
315    #[test]
316    fn sends_a_valid_ping_envelope() {
317        let (endpoint, server) = server_once(
318            "{\"v\":1,\"ok\":true,\"result\":{\"version\":\"0.4.33\",\"ipc_v\":1,\"capabilities\":{}}}\n",
319        );
320        let ping = CCCCClient::new(endpoint).ping().expect("ping");
321        assert_eq!(ping.version, "0.4.33");
322        assert_eq!(ping.ipc_v, 1);
323
324        let request: Value =
325            serde_json::from_str(&server.join().expect("server thread")).expect("request JSON");
326        assert_eq!(request, json!({"v": 1, "op": "ping", "args": {}}));
327    }
328
329    #[test]
330    fn preserves_structured_daemon_errors() {
331        let (endpoint, server) = server_once(
332            "{\"v\":1,\"ok\":false,\"result\":{},\"error\":{\"code\":\"unknown_op\",\"message\":\"unknown\",\"details\":{}}}\n",
333        );
334        let error = CCCCClient::new(endpoint)
335            .call("future_op", Map::new())
336            .expect_err("daemon error");
337        assert!(matches!(
338            error,
339            Error::Daemon(crate::DaemonError { ref code, .. }) if code == "unknown_op"
340        ));
341        server.join().expect("server thread");
342    }
343
344    #[test]
345    fn never_probes_destructive_or_streaming_operations() {
346        assert!(operation_probe_is_unsafe("shutdown"));
347        assert!(operation_probe_is_unsafe("term_attach"));
348        assert!(!operation_probe_is_unsafe("group_show"));
349    }
350}
cccc_sdk/client.rs

cccc_sdk/
client.rs
