use std::collections::BTreeMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
use std::time::Duration;
use serde::de::DeserializeOwned;
use serde_json::{json, Map, Value};
use crate::{
discover_endpoint, ContextDetail, DaemonEndpoint, DaemonRequest, DaemonResponse, Error,
MessageHistoryMode, MessageMode, PingResult, ReplyMessageMode, Result, TerminalHistoryOptions,
TerminalHistoryResult, TerminalResizeResult, TerminalSinceOptions, TerminalSinceResult,
TerminalSnapshotOptions, TerminalSnapshotResult, WebModelDeliveryMode,
WebModelDeliveryPreferencesResult, WebModelRuntimeRecoverTurnResult,
};
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_REQUEST_BYTES: usize = 2_000_000;
const MAX_RESPONSE_BYTES: usize = 4_000_000;
enum CallAttemptError {
Connect(Error),
Exchange(Error),
}
#[derive(Clone, Debug, Default)]
pub struct CompatibilityRequirements<'a> {
pub minimum_ipc_version: u32,
pub capabilities: BTreeMap<&'a str, bool>,
pub operations: Vec<&'a str>,
}
#[derive(Clone, Debug)]
pub struct CCCCClient {
endpoint: DaemonEndpoint,
rediscovered_endpoint: Arc<RwLock<Option<DaemonEndpoint>>>,
timeout: Duration,
discovery_home: Option<PathBuf>,
rediscover_on_connect_failure: bool,
}
impl CCCCClient {
pub fn discover() -> Result<Self> {
Self::discover_in(None)
}
pub fn discover_in(cccc_home: Option<&Path>) -> Result<Self> {
Ok(Self {
endpoint: discover_endpoint(cccc_home)?,
rediscovered_endpoint: Arc::new(RwLock::new(None)),
timeout: DEFAULT_TIMEOUT,
discovery_home: cccc_home.map(Path::to_path_buf),
rediscover_on_connect_failure: true,
})
}
pub fn new(endpoint: DaemonEndpoint) -> Self {
Self {
endpoint,
rediscovered_endpoint: Arc::new(RwLock::new(None)),
timeout: DEFAULT_TIMEOUT,
discovery_home: None,
rediscover_on_connect_failure: false,
}
}
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}
pub fn endpoint(&self) -> &DaemonEndpoint {
&self.endpoint
}
pub fn current_endpoint(&self) -> DaemonEndpoint {
self.rediscovered_endpoint
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone()
.unwrap_or_else(|| self.endpoint.clone())
}
pub fn call(&self, op: &str, args: Map<String, Value>) -> Result<Map<String, Value>> {
let response = self.call_raw(op, args)?;
if response.v != 1 {
return Err(Error::UnsupportedIpcVersion(response.v));
}
if response.ok {
return Ok(response.result);
}
Err(Error::Daemon(response.error.unwrap_or_else(|| {
crate::DaemonError {
code: "error".into(),
message: "daemon returned ok=false without an error".into(),
details: json!({}),
}
})))
}
pub fn call_raw(&self, op: &str, args: Map<String, Value>) -> Result<DaemonResponse> {
if op.trim().is_empty() {
return Err(Error::Incompatible("operation name cannot be empty".into()));
}
let request = DaemonRequest {
v: 1,
op,
args: &args,
};
let mut encoded = serde_json::to_vec(&request)?;
encoded.push(b'\n');
if encoded.len() > MAX_REQUEST_BYTES {
return Err(Error::RequestTooLarge(MAX_REQUEST_BYTES));
}
let endpoint = self.current_endpoint();
let response = match self.call_at(&endpoint, &encoded) {
Ok(response) => Ok(response),
Err(CallAttemptError::Connect(_)) if self.rediscover_on_connect_failure => {
let endpoint = discover_endpoint(self.discovery_home.as_deref())?;
*self
.rediscovered_endpoint
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(endpoint.clone());
self.call_at(&endpoint, &encoded)
.map_err(|retry_error| finish_attempt_error(op, retry_error))
}
Err(error) => Err(finish_attempt_error(op, error)),
}?;
if response.v != 1 {
return Err(Error::UnsupportedIpcVersion(response.v));
}
Ok(response)
}
fn call_at(
&self,
endpoint: &DaemonEndpoint,
encoded: &[u8],
) -> std::result::Result<DaemonResponse, CallAttemptError> {
match endpoint {
DaemonEndpoint::Tcp { host, port } => {
let address = (host.as_str(), *port)
.to_socket_addrs()
.map_err(|error| CallAttemptError::Connect(error.into()))?
.next()
.ok_or_else(|| Error::InvalidEndpoint(format!("cannot resolve {host}:{port}")))
.map_err(CallAttemptError::Connect)?;
let mut stream = TcpStream::connect_timeout(&address, self.timeout)
.map_err(|error| CallAttemptError::Connect(error.into()))?;
stream
.set_read_timeout(Some(self.timeout))
.map_err(|error| CallAttemptError::Connect(error.into()))?;
stream
.set_write_timeout(Some(self.timeout))
.map_err(|error| CallAttemptError::Connect(error.into()))?;
exchange(&mut stream, encoded).map_err(CallAttemptError::Exchange)
}
DaemonEndpoint::Unix(path) => self.call_unix(path, encoded),
}
}
#[cfg(unix)]
fn call_unix(
&self,
path: &Path,
encoded: &[u8],
) -> std::result::Result<DaemonResponse, CallAttemptError> {
use std::os::unix::net::UnixStream;
let mut stream =
UnixStream::connect(path).map_err(|error| CallAttemptError::Connect(error.into()))?;
stream
.set_read_timeout(Some(self.timeout))
.map_err(|error| CallAttemptError::Connect(error.into()))?;
stream
.set_write_timeout(Some(self.timeout))
.map_err(|error| CallAttemptError::Connect(error.into()))?;
exchange(&mut stream, encoded).map_err(CallAttemptError::Exchange)
}
#[cfg(not(unix))]
fn call_unix(
&self,
_path: &Path,
_encoded: &[u8],
) -> std::result::Result<DaemonResponse, CallAttemptError> {
Err(CallAttemptError::Connect(Error::UnixSocketUnsupported))
}
pub fn ping(&self) -> Result<PingResult> {
self.call_typed("ping", Map::new())
}
pub fn groups(&self) -> Result<Map<String, Value>> {
self.call("groups", Map::new())
}
pub fn group_show(&self, group_id: &str) -> Result<Map<String, Value>> {
self.call("group_show", object([("group_id", json!(group_id))]))
}
pub fn send(
&self,
group_id: &str,
text: &str,
message_mode: MessageMode,
by: &str,
) -> Result<Map<String, Value>> {
self.call(
"send",
object([
("group_id", json!(group_id)),
("text", json!(text)),
("message_mode", json!(message_mode.as_str())),
("by", json!(by)),
]),
)
}
pub fn reply(
&self,
group_id: &str,
reply_to: &str,
text: &str,
by: &str,
) -> Result<Map<String, Value>> {
self.reply_with_mode(group_id, reply_to, text, ReplyMessageMode::Send, by)
}
pub fn reply_with_mode(
&self,
group_id: &str,
reply_to: &str,
text: &str,
message_mode: ReplyMessageMode,
by: &str,
) -> Result<Map<String, Value>> {
self.call(
"reply",
object([
("group_id", json!(group_id)),
("reply_to", json!(reply_to)),
("text", json!(text)),
("message_mode", json!(message_mode.as_str())),
("by", json!(by)),
]),
)
}
pub fn reply_request_cancel(
&self,
group_id: &str,
source_event_id: &str,
by: &str,
) -> Result<Map<String, Value>> {
self.call(
"reply_request_cancel",
object([
("group_id", json!(group_id)),
("source_event_id", json!(source_event_id)),
("by", json!(by)),
]),
)
}
pub fn message_deliver(
&self,
group_id: &str,
source_event_id: &str,
actor_ids: &[&str],
by: &str,
force_ambiguous: bool,
) -> Result<Map<String, Value>> {
if actor_ids.is_empty() || actor_ids.iter().any(|actor_id| actor_id.trim().is_empty()) {
return Err(Error::InvalidArgument(
"message_deliver requires one or more non-empty actor_ids".into(),
));
}
self.call(
"message_deliver",
object([
("group_id", json!(group_id)),
("source_event_id", json!(source_event_id)),
("actor_ids", json!(actor_ids)),
("by", json!(by)),
("force_ambiguous", json!(force_ambiguous)),
]),
)
}
pub fn inbox_peek(
&self,
group_id: &str,
actor_id: &str,
by: &str,
limit: Option<u32>,
) -> Result<Map<String, Value>> {
let mut args = object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("by", json!(by)),
]);
if let Some(limit) = limit {
args.insert("limit".into(), json!(limit));
}
self.call("inbox_peek", args)
}
pub fn inbox_read(
&self,
group_id: &str,
actor_id: &str,
by: &str,
limit: Option<u32>,
) -> Result<Map<String, Value>> {
let mut args = object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("by", json!(by)),
]);
if let Some(limit) = limit {
args.insert("limit".into(), json!(limit));
}
self.call("inbox_read", args)
}
#[allow(clippy::too_many_arguments)]
pub fn message_history(
&self,
group_id: &str,
actor_id: &str,
by: &str,
mode: MessageHistoryMode,
query: Option<&str>,
before_event_id: Option<&str>,
limit: Option<u32>,
) -> Result<Map<String, Value>> {
let mut args = object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("by", json!(by)),
("mode", json!(mode.as_str())),
]);
if let Some(query) = query.filter(|value| !value.is_empty()) {
args.insert("query".into(), json!(query));
}
if let Some(before_event_id) = before_event_id.filter(|value| !value.is_empty()) {
args.insert("before_event_id".into(), json!(before_event_id));
}
if let Some(limit) = limit {
args.insert("limit".into(), json!(limit));
}
self.call("message_history", args)
}
pub fn context_get(&self, group_id: &str) -> Result<Map<String, Value>> {
self.context_get_with_detail(group_id, ContextDetail::Full)
}
pub fn context_get_with_detail(
&self,
group_id: &str,
detail: ContextDetail,
) -> Result<Map<String, Value>> {
self.call(
"context_get",
object([
("group_id", json!(group_id)),
("detail", json!(detail.as_str())),
]),
)
}
pub fn context_sync(
&self,
group_id: &str,
by: &str,
operations: Vec<Value>,
) -> Result<Map<String, Value>> {
self.context_sync_checked(group_id, by, operations, None, false)
}
pub fn context_sync_checked(
&self,
group_id: &str,
by: &str,
operations: Vec<Value>,
if_version: Option<&str>,
dry_run: bool,
) -> Result<Map<String, Value>> {
let mut args = object([
("group_id", json!(group_id)),
("by", json!(by)),
("ops", Value::Array(operations)),
("dry_run", json!(dry_run)),
]);
if let Some(version) = if_version {
args.insert("if_version".into(), json!(version));
}
self.call("context_sync", args)
}
pub fn terminal_history(
&self,
group_id: &str,
actor_id: &str,
options: &TerminalHistoryOptions,
) -> Result<TerminalHistoryResult> {
let mut args = object([("group_id", json!(group_id)), ("actor_id", json!(actor_id))]);
if let Some(render_before) = options.render_before {
args.insert("render_before".into(), json!(render_before));
}
if let Some(before) = options.before {
args.insert("before".into(), json!(before));
}
if let Some(limit_bytes) = options.limit_bytes {
args.insert("limit_bytes".into(), json!(limit_bytes));
}
if let Some(strip_ansi) = options.strip_ansi {
args.insert("strip_ansi".into(), json!(strip_ansi));
}
if let Some(compact) = options.compact {
args.insert("compact".into(), json!(compact));
}
if let Some(by) = options.by.as_deref() {
args.insert("by".into(), json!(by));
}
self.call_typed("terminal_history", args)
}
pub fn terminal_since(
&self,
group_id: &str,
actor_id: &str,
after: u64,
options: &TerminalSinceOptions,
) -> Result<TerminalSinceResult> {
let mut args = object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("after", json!(after)),
]);
if let Some(limit_bytes) = options.limit_bytes {
args.insert("limit_bytes".into(), json!(limit_bytes));
}
if let Some(by) = options.by.as_deref() {
args.insert("by".into(), json!(by));
}
self.call_typed("terminal_since", args)
}
pub fn terminal_snapshot(
&self,
group_id: &str,
actor_id: &str,
options: &TerminalSnapshotOptions,
) -> Result<TerminalSnapshotResult> {
let mut args = object([("group_id", json!(group_id)), ("actor_id", json!(actor_id))]);
if let Some(limit_bytes) = options.limit_bytes {
args.insert("limit_bytes".into(), json!(limit_bytes));
}
if let Some(by) = options.by.as_deref() {
args.insert("by".into(), json!(by));
}
self.call_typed("terminal_snapshot", args)
}
pub fn term_resize(
&self,
group_id: &str,
actor_id: &str,
cols: u32,
rows: u32,
) -> Result<TerminalResizeResult> {
let args = object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("cols", json!(cols)),
("rows", json!(rows)),
]);
match self.call_typed("term_resize", args.clone()) {
Err(Error::Daemon(error)) if error.code == "unknown_op" => {
let legacy = self.call("terminal_resize", args)?;
let resolved_cols = legacy
.get("cols")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(cols);
let resolved_rows = legacy
.get("rows")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(rows);
Ok(TerminalResizeResult {
group_id: group_id.to_owned(),
actor_id: actor_id.to_owned(),
cols: resolved_cols,
rows: resolved_rows,
})
}
result => result,
}
}
pub fn web_model_delivery_preferences_get(
&self,
group_id: &str,
actor_id: &str,
) -> Result<WebModelDeliveryPreferencesResult> {
self.call_typed(
"web_model_delivery_preferences_get",
object([("group_id", json!(group_id)), ("actor_id", json!(actor_id))]),
)
}
pub fn web_model_delivery_preferences_update(
&self,
group_id: &str,
actor_id: &str,
mode: WebModelDeliveryMode,
by: &str,
) -> Result<WebModelDeliveryPreferencesResult> {
self.call_typed(
"web_model_delivery_preferences_update",
object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("mode", json!(mode)),
("by", json!(by)),
]),
)
}
pub fn web_model_runtime_recover_turn(
&self,
group_id: &str,
actor_id: &str,
event_ids: &[String],
) -> Result<WebModelRuntimeRecoverTurnResult> {
if event_ids.is_empty() || event_ids.iter().any(|event_id| event_id.trim().is_empty()) {
return Err(Error::InvalidArgument(
"event_ids must contain at least one non-empty event id".into(),
));
}
self.call_typed(
"web_model_runtime_recover_turn",
object([
("group_id", json!(group_id)),
("actor_id", json!(actor_id)),
("event_ids", json!(event_ids)),
]),
)
}
pub fn assert_compatible(
&self,
requirements: &CompatibilityRequirements<'_>,
) -> Result<PingResult> {
let ping = self.ping()?;
let minimum = requirements.minimum_ipc_version.max(1);
if ping.ipc_v < minimum {
return Err(Error::Incompatible(format!(
"daemon ipc_v={} is below required ipc_v={minimum}",
ping.ipc_v
)));
}
for (name, required) in &requirements.capabilities {
if *required && ping.capabilities.get(*name).and_then(Value::as_bool) != Some(true) {
return Err(Error::Incompatible(format!(
"daemon capability {name}=true is required"
)));
}
}
for operation in &requirements.operations {
if *operation == "ping" {
continue;
}
let advertised = ping.capabilities.get(*operation).and_then(Value::as_bool);
if advertised == Some(false) {
return Err(Error::Incompatible(format!(
"daemon does not support operation {operation}"
)));
}
if !operation_probe_is_safe(operation) {
if advertised == Some(true) {
continue;
}
return Err(Error::Incompatible(format!("cannot safely verify operation {operation}; no advertised capability or safe probe")));
}
match self.call(operation, Map::new()) {
Err(Error::Daemon(error)) if error.code == "unknown_op" => {
if *operation == "term_resize" {
match self.call("terminal_resize", Map::new()) {
Err(Error::Daemon(alias_error)) if alias_error.code == "unknown_op" => {
}
Err(Error::Daemon(_)) | Ok(_) => continue,
Err(error) => return Err(error),
}
}
return Err(Error::Incompatible(format!(
"daemon does not support operation {operation}"
)));
}
Err(Error::Daemon(_)) | Ok(_) => {}
Err(error) => return Err(error),
}
}
Ok(ping)
}
fn call_typed<T: DeserializeOwned>(&self, op: &str, args: Map<String, Value>) -> Result<T> {
let result = self.call(op, args)?;
Ok(serde_json::from_value(Value::Object(result))?)
}
}
fn operation_probe_is_safe(operation: &str) -> bool {
matches!(
operation,
"groups"
| "group_show"
| "group_preamble_get"
| "group_preamble_set"
| "group_preamble_reset"
| "send"
| "tracked_send"
| "send_files"
| "reply"
| "inbox_peek"
| "inbox_read"
| "context_get"
| "context_sync"
| "message_deliver"
| "message_history"
| "reply_request_cancel"
| "send_cross_group"
| "memory_search"
| "memory_get"
| "memory_write"
| "memory_profile_get"
| "memory_health"
| "actor_new_session"
| "group_reset"
| "group_copy_export_file"
| "terminal_history"
| "terminal_since"
| "terminal_snapshot"
| "term_resize"
| "web_model_delivery_preferences_get"
| "web_model_delivery_preferences_update"
| "web_model_runtime_recover_turn"
| "assistant_voice_document_library"
| "assistant_voice_document_library_update"
| "assistant_voice_document_delete"
| "events_stream"
| "connect_catalog"
| "connect_send"
| "connect_send_files"
)
}
fn finish_attempt_error(op: &str, error: CallAttemptError) -> Error {
match error {
CallAttemptError::Connect(error) => error,
CallAttemptError::Exchange(error) => Error::OutcomeUnknown {
op: op.to_owned(),
message: error.to_string(),
},
}
}
fn object<const N: usize>(entries: [(&str, Value); N]) -> Map<String, Value> {
entries
.into_iter()
.map(|(key, value)| (key.to_owned(), value))
.collect()
}
fn exchange<S: Read + Write>(stream: &mut S, request: &[u8]) -> Result<DaemonResponse> {
stream.write_all(request)?;
stream.flush()?;
let mut reader = BufReader::new(stream);
let mut response = Vec::new();
let read = reader
.by_ref()
.take((MAX_RESPONSE_BYTES + 1) as u64)
.read_until(b'\n', &mut response)?;
if read == 0 {
return Err(Error::EmptyResponse);
}
if response.len() > MAX_RESPONSE_BYTES {
return Err(Error::ResponseTooLarge(MAX_RESPONSE_BYTES));
}
Ok(serde_json::from_slice(&response)?)
}
#[cfg(test)]
mod tests {
use super::*;
use std::env;
use std::fs;
use std::net::TcpListener;
use std::thread;
use std::time::{SystemTime, UNIX_EPOCH};
fn server_once(response: &'static str) -> (DaemonEndpoint, thread::JoinHandle<String>) {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server");
let address = listener.local_addr().expect("local address");
let handle = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept client");
let mut request = String::new();
BufReader::new(&mut stream)
.read_line(&mut request)
.expect("read request");
stream
.write_all(response.as_bytes())
.expect("write response");
request
});
(
DaemonEndpoint::Tcp {
host: "127.0.0.1".into(),
port: address.port(),
},
handle,
)
}
fn server_sequence(
responses: Vec<&'static str>,
) -> (DaemonEndpoint, thread::JoinHandle<Vec<String>>) {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server");
let address = listener.local_addr().expect("local address");
let handle = thread::spawn(move || {
responses
.into_iter()
.map(|response| {
let (mut stream, _) = listener.accept().expect("accept client");
let mut request = String::new();
BufReader::new(&mut stream)
.read_line(&mut request)
.expect("read request");
stream
.write_all(response.as_bytes())
.expect("write response");
request
})
.collect()
});
(
DaemonEndpoint::Tcp {
host: "127.0.0.1".into(),
port: address.port(),
},
handle,
)
}
fn temp_home(label: &str) -> PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_nanos();
env::temp_dir().join(format!(
"cccc-sdk-client-{label}-{}-{nonce}",
std::process::id()
))
}
fn write_tcp_descriptor(home: &Path, port: u16) {
let daemon = home.join("daemon");
fs::create_dir_all(&daemon).expect("create daemon directory");
fs::write(
daemon.join("ccccd.addr.json"),
format!(r#"{{"v":1,"transport":"tcp","host":"127.0.0.1","port":{port}}}"#),
)
.expect("write descriptor");
}
#[test]
fn sends_a_valid_ping_envelope() {
let (endpoint, server) = server_once(
"{\"v\":1,\"ok\":true,\"result\":{\"version\":\"0.4.33\",\"implementation\":\"rust\",\"ipc_v\":1,\"capabilities\":{}}}\n",
);
let ping = CCCCClient::new(endpoint).ping().expect("ping");
assert_eq!(ping.version, "0.4.33");
assert_eq!(ping.ipc_v, 1);
let request: Value =
serde_json::from_str(&server.join().expect("server thread")).expect("request JSON");
assert_eq!(request, json!({"v": 1, "op": "ping", "args": {}}));
}
#[test]
fn voice_library_preserves_actions_and_empty_values_on_the_wire() {
use crate::VoiceDocumentLibraryAction as Action;
let cases = [
(
Action::CreateFolder {
name: "Meetings".into(),
},
json!({"action":"create_folder","name":"Meetings"}),
),
(
Action::RenameFolder {
folder_id: "f_1".into(),
name: "Archive".into(),
},
json!({"action":"rename_folder","folder_id":"f_1","name":"Archive"}),
),
(
Action::RemoveFolder {
folder_id: "f_1".into(),
},
json!({"action":"remove_folder","folder_id":"f_1"}),
),
(
Action::ReorderRoot {
root_order: vec!["folder:f_1".into(), "document:notes.md".into()],
},
json!({"action":"reorder_root","root_order":["folder:f_1","document:notes.md"]}),
),
(
Action::ReorderRoot { root_order: vec![] },
json!({"action":"reorder_root","root_order":[]}),
),
(
Action::Rename {
document_path: "notes.md".into(),
name: "Summary".into(),
},
json!({"action":"rename","document_path":"notes.md","name":"Summary"}),
),
(
Action::Move {
document_path: "notes.md".into(),
folder_id: "".into(),
},
json!({"action":"move","document_path":"notes.md","folder_id":""}),
),
(
Action::Restore {
document_path: "notes.md".into(),
},
json!({"action":"restore","document_path":"notes.md"}),
),
];
let response = "{\"v\":1,\"ok\":true,\"result\":{\"folders\":[],\"root_order\":[],\"documents\":[{\"status\":\"archived\"}]}}\n";
let (endpoint, server) = server_sequence(vec![response; cases.len() + 1]);
let client = CCCCClient::new(endpoint);
let library = client
.assistant_voice_document_library("g_1")
.expect("library");
assert_eq!(library.documents[0]["status"], "archived");
for (action, _) in &cases {
client
.assistant_voice_document_library_update("g_1", action, Some("foreman"))
.expect("update");
}
let requests: Vec<Value> = server
.join()
.expect("server")
.iter()
.map(|line| serde_json::from_str(line).expect("JSON"))
.collect();
assert_eq!(
requests[0],
json!({"v":1,"op":"assistant_voice_document_library","args":{"group_id":"g_1"}})
);
for (index, (_, mut expected)) in cases.into_iter().enumerate() {
expected["group_id"] = json!("g_1");
expected["by"] = json!("foreman");
assert_eq!(
requests[index + 1],
json!({"v":1,"op":"assistant_voice_document_library_update","args":expected})
);
}
}
#[test]
fn voice_delete_preserves_recording_error() {
let (endpoint, server) = server_once("{\"v\":1,\"ok\":false,\"error\":{\"code\":\"voice_recording_active\",\"message\":\"Stop recording\"}}\n");
let error = CCCCClient::new(endpoint)
.assistant_voice_document_delete("g_1", "notes.md", None)
.expect_err("recording");
assert!(
matches!(error, Error::Daemon(crate::DaemonError { ref code, .. }) if code == "voice_recording_active")
);
let request: Value = serde_json::from_str(&server.join().expect("server")).expect("JSON");
assert_eq!(
request,
json!({"v":1,"op":"assistant_voice_document_delete","args":{"group_id":"g_1","document_path":"notes.md","by":"user"}})
);
}
#[test]
fn preserves_structured_daemon_errors() {
let (endpoint, server) = server_once(
"{\"v\":1,\"ok\":false,\"result\":{},\"error\":{\"code\":\"unknown_op\",\"message\":\"unknown\",\"details\":{}}}\n",
);
let error = CCCCClient::new(endpoint)
.call("future_op", Map::new())
.expect_err("daemon error");
assert!(matches!(
error,
Error::Daemon(crate::DaemonError { ref code, .. }) if code == "unknown_op"
));
server.join().expect("server thread");
}
#[test]
fn maps_current_message_delivery_and_inbox_operations() {
let response = "{\"v\":1,\"ok\":true,\"result\":{}}\n";
let (endpoint, server) = server_sequence(vec![response; 7]);
let client = CCCCClient::new(endpoint);
client
.send("g_1", "FYI", MessageMode::Mail, "user")
.expect("send Mail");
client
.reply_with_mode(
"g_1",
"e_request",
"quiet follow-up",
ReplyMessageMode::Mail,
"peer-1",
)
.expect("reply with Mail");
client
.reply_request_cancel("g_1", "e_request", "user")
.expect("cancel reply request");
client
.message_deliver("g_1", "e_mail", &["peer-1"], "user", true)
.expect("deliver existing message");
client
.inbox_peek("g_1", "peer-1", "user", Some(5))
.expect("peek Inbox");
client
.inbox_read("g_1", "peer-1", "peer-1", Some(3))
.expect("read Inbox");
client
.message_history(
"g_1",
"peer-1",
"user",
MessageHistoryMode::Send,
Some("decision"),
Some("e_before"),
Some(7),
)
.expect("read message history");
let requests: Vec<Value> = server
.join()
.expect("server thread")
.iter()
.map(|request| serde_json::from_str(request).expect("request JSON"))
.collect();
assert_eq!(requests[0]["args"]["message_mode"], "mail");
assert_eq!(requests[1]["op"], "reply");
assert_eq!(requests[1]["args"]["message_mode"], "mail");
assert_eq!(requests[2]["op"], "reply_request_cancel");
assert_eq!(requests[3]["args"]["actor_ids"], json!(["peer-1"]));
assert_eq!(requests[3]["args"]["force_ambiguous"], true);
assert_eq!(requests[4]["op"], "inbox_peek");
assert_eq!(requests[5]["op"], "inbox_read");
assert_eq!(requests[6]["op"], "message_history");
assert_eq!(requests[6]["args"]["mode"], "send");
assert_eq!(requests[6]["args"]["query"], "decision");
assert_eq!(requests[6]["args"]["before_event_id"], "e_before");
assert_eq!(requests[6]["args"]["limit"], 7);
assert!(matches!(
client.message_deliver("g_1", "e_mail", &[], "user", false),
Err(Error::InvalidArgument(_))
));
}
#[test]
fn context_get_uses_the_current_projection_contract() {
let (endpoint, server) = server_once("{\"v\":1,\"ok\":true,\"result\":{}}\n");
CCCCClient::new(endpoint)
.context_get_with_detail("g_1", ContextDetail::Summary)
.expect("summary context");
let request: Value =
serde_json::from_str(&server.join().expect("server thread")).expect("request JSON");
assert_eq!(
request,
json!({"v": 1, "op": "context_get", "args": {"group_id": "g_1", "detail": "summary"}})
);
}
#[test]
fn never_probes_destructive_or_streaming_operations() {
assert!(!operation_probe_is_safe("shutdown"));
assert!(!operation_probe_is_safe("group_create"));
assert!(!operation_probe_is_safe("remote_access_start"));
assert!(!operation_probe_is_safe("term_attach"));
assert!(operation_probe_is_safe("group_show"));
}
#[test]
fn call_raw_rejects_an_unsupported_response_version() {
let (endpoint, server) = server_once("{\"v\":2,\"ok\":true,\"result\":{}}\n");
let error = CCCCClient::new(endpoint)
.call_raw("ping", Map::new())
.expect_err("unsupported version");
assert!(matches!(error, Error::UnsupportedIpcVersion(2)));
server.join().expect("server thread");
}
#[test]
fn maps_current_terminal_and_web_model_operations() {
let preference = "{\"v\":1,\"ok\":true,\"result\":{\"group_id\":\"g_1\",\"actor_id\":\"web-1\",\"preference\":{\"mode\":\"image_compat\",\"updated_at\":\"now\",\"updated_by\":\"user\"}}}\n";
let (endpoint, server) = server_sequence(vec![
"{\"v\":1,\"ok\":true,\"result\":{\"data\":\"screen\",\"start_cursor\":1,\"end_cursor\":7}}\n",
preference,
preference,
"{\"v\":1,\"ok\":true,\"result\":{\"status\":\"recovered\",\"turn\":{\"turn_id\":\"turn-1\",\"group_id\":\"g_1\",\"actor_id\":\"web-1\",\"event_ids\":[\"e_1\"],\"latest_event_id\":\"e_1\",\"latest_ts\":\"now\",\"messages\":[],\"coalesced_text\":\"hello\",\"system_prompt\":\"system\",\"delivery\":{\"mode\":\"recovery_no_cursor_mutation\",\"cursor_committed\":true,\"web_model_mode\":\"image_compat\"}}}}\n",
]);
let client = CCCCClient::new(endpoint);
let snapshot = client
.terminal_snapshot(
"g_1",
"web-1",
&TerminalSnapshotOptions {
limit_bytes: Some(4096),
by: Some("user".into()),
},
)
.expect("terminal snapshot");
assert_eq!(snapshot.end_cursor, 7);
let preference = client
.web_model_delivery_preferences_get("g_1", "web-1")
.expect("preference get");
assert_eq!(
preference.preference.mode,
WebModelDeliveryMode::ImageCompat
);
client
.web_model_delivery_preferences_update(
"g_1",
"web-1",
WebModelDeliveryMode::ImageCompat,
"user",
)
.expect("preference update");
let recovered = client
.web_model_runtime_recover_turn("g_1", "web-1", &["e_1".into()])
.expect("recover turn");
assert_eq!(recovered.status, "recovered");
let requests: Vec<Value> = server
.join()
.expect("server thread")
.iter()
.map(|request| serde_json::from_str(request).expect("request JSON"))
.collect();
assert_eq!(
requests
.iter()
.map(|request| request["op"].as_str().expect("op"))
.collect::<Vec<_>>(),
vec![
"terminal_snapshot",
"web_model_delivery_preferences_get",
"web_model_delivery_preferences_update",
"web_model_runtime_recover_turn",
]
);
assert_eq!(requests[0]["args"]["limit_bytes"], 4096);
assert_eq!(requests[2]["args"]["mode"], "image_compat");
assert_eq!(requests[3]["args"]["event_ids"], json!(["e_1"]));
}
#[test]
fn term_resize_prefers_the_standard_op_and_falls_back_for_unknown_op() {
let (endpoint, server) = server_sequence(vec![
"{\"v\":1,\"ok\":false,\"result\":{},\"error\":{\"code\":\"unknown_op\",\"message\":\"unknown\",\"details\":{}}}\n",
"{\"v\":1,\"ok\":true,\"result\":{\"resized\":true,\"cols\":120,\"rows\":40}}\n",
]);
let resized = CCCCClient::new(endpoint)
.term_resize("g_1", "a_1", 120, 40)
.expect("resize");
assert_eq!(resized.group_id, "g_1");
assert_eq!(resized.actor_id, "a_1");
assert_eq!(resized.cols, 120);
let requests = server.join().expect("server thread");
let operations: Vec<String> = requests
.iter()
.map(|request| {
serde_json::from_str::<Value>(request).expect("request JSON")["op"]
.as_str()
.expect("op")
.to_owned()
})
.collect();
assert_eq!(operations, vec!["term_resize", "terminal_resize"]);
}
#[test]
fn compatibility_probe_accepts_the_bounded_resize_alias() {
let (endpoint, server) = server_sequence(vec![
"{\"v\":1,\"ok\":true,\"result\":{\"version\":\"test\",\"implementation\":\"rust\",\"ipc_v\":1,\"capabilities\":{}}}\n",
"{\"v\":1,\"ok\":false,\"result\":{},\"error\":{\"code\":\"unknown_op\",\"message\":\"unknown\",\"details\":{}}}\n",
"{\"v\":1,\"ok\":false,\"result\":{},\"error\":{\"code\":\"invalid_request\",\"message\":\"missing args\",\"details\":{}}}\n",
]);
let requirements = CompatibilityRequirements {
minimum_ipc_version: 1,
operations: vec!["term_resize"],
..Default::default()
};
CCCCClient::new(endpoint)
.assert_compatible(&requirements)
.expect("legacy alias is usable through the SDK");
let operations: Vec<String> = server
.join()
.expect("server thread")
.iter()
.map(|request| {
serde_json::from_str::<Value>(request).expect("request JSON")["op"]
.as_str()
.expect("op")
.to_owned()
})
.collect();
assert_eq!(operations, vec!["ping", "term_resize", "terminal_resize"]);
}
#[test]
fn unknown_or_side_effectful_probes_fail_without_dispatch() {
for op in [
"connect_direct_configure",
"membership_logout",
"future_mutation",
] {
let (endpoint, server) = server_sequence(vec![
"{\"v\":1,\"ok\":true,\"result\":{\"implementation\":\"rust\",\"ipc_v\":1,\"capabilities\":{}}}\n",
]);
let requirements = CompatibilityRequirements {
operations: vec![op],
..Default::default()
};
let error = CCCCClient::new(endpoint)
.assert_compatible(&requirements)
.unwrap_err();
assert!(matches!(error, Error::Incompatible(_)));
assert!(error.to_string().contains("cannot safely verify"));
assert_eq!(server.join().unwrap().len(), 1);
}
}
#[test]
fn connect_and_checked_context_keep_identity_and_conflict_tokens() {
let (endpoint, server) = server_sequence(vec![
"{\"v\":1,\"ok\":true,\"result\":{\"accepted\":true}}\n",
"{\"v\":1,\"ok\":false,\"error\":{\"code\":\"version_conflict\",\"message\":\"changed\",\"details\":{}}}\n",
]);
let client = CCCCClient::new(endpoint);
let mut options = crate::ConnectSendOptions {
group_id: "local".into(),
instance_id: "peer".into(),
target_group_id: "remote".into(),
client_id: "key".into(),
text: "hello".into(),
message_mode: MessageMode::Mail,
by: None,
to: None,
format: None,
insight: None,
attachments: None,
};
client.connect_send(&options).unwrap();
let error = client
.context_sync_checked("local", "user", vec![], Some("version"), false)
.unwrap_err();
assert!(matches!(error, Error::Daemon(ref e) if e.code == "version_conflict"));
for key in ["".to_owned(), "界".repeat(43)] {
options.client_id = key;
assert!(matches!(
client.connect_send(&options),
Err(Error::InvalidArgument(_))
));
}
let calls: Vec<Value> = server
.join()
.unwrap()
.iter()
.map(|line| serde_json::from_str(line).unwrap())
.collect();
assert_eq!(calls[0]["args"]["instance_id"], "peer");
assert_eq!(calls[0]["args"]["target_group_id"], "remote");
assert_eq!(calls[0]["args"]["client_id"], "key");
assert_eq!(calls[0]["args"]["message_mode"], "mail");
assert_eq!(calls[1]["args"]["if_version"], "version");
}
#[test]
fn rediscovers_endpoint_after_connect_failure_before_exchange() {
let home = temp_home("rediscover");
let stale = TcpListener::bind("127.0.0.1:0").expect("bind stale port");
let stale_port = stale.local_addr().expect("stale address").port();
drop(stale);
write_tcp_descriptor(&home, stale_port);
let client = CCCCClient::discover_in(Some(&home)).expect("discover stale endpoint");
let listener = TcpListener::bind("127.0.0.1:0").expect("bind live server");
let live_port = listener.local_addr().expect("live address").port();
write_tcp_descriptor(&home, live_port);
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept client");
let mut request = String::new();
BufReader::new(&mut stream)
.read_line(&mut request)
.expect("read request");
stream
.write_all(b"{\"v\":1,\"ok\":true,\"result\":{\"implementation\":\"rust\",\"ipc_v\":1,\"capabilities\":{}}}\n")
.expect("write response");
});
assert_eq!(client.ping().expect("ping").ipc_v, 1);
assert_eq!(
client.current_endpoint(),
DaemonEndpoint::Tcp {
host: "127.0.0.1".into(),
port: live_port,
}
);
server.join().expect("server thread");
fs::remove_dir_all(home).expect("cleanup");
}
#[test]
fn exchange_failure_is_reported_as_outcome_unknown_without_replay() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server");
let address = listener.local_addr().expect("local address");
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept client");
let mut request = String::new();
BufReader::new(&mut stream)
.read_line(&mut request)
.expect("read request");
});
let client = CCCCClient::new(DaemonEndpoint::Tcp {
host: "127.0.0.1".into(),
port: address.port(),
});
assert!(matches!(
client.call("non_idempotent_write", Map::new()),
Err(Error::OutcomeUnknown { ref op, .. }) if op == "non_idempotent_write"
));
server.join().expect("server thread");
}
}