rx4/
ipc.rs

1//! JSON-RPC IPC server over Unix socket.
2
3use crate::agent::{Agent, ToolRegistry};
4use crate::plugin::PluginRegistry;
5use crate::session::Session;
6use serde_json::Value;
7use std::os::unix::net::UnixListener;
8use std::path::Path;
9use std::sync::{Arc, Mutex};
10use subtle::ConstantTimeEq;
11use tokio::sync::Mutex as AsyncMutex;
12use tracing::{info, warn};
13
14pub struct IpcServer {
15    pub socket_path: String,
16    pub agent: Arc<AsyncMutex<Agent>>,
17    pub tools: Arc<Mutex<ToolRegistry>>,
18    pub plugins: Arc<Mutex<PluginRegistry>>,
19    pub session: Arc<Mutex<Session>>,
20}
21
22impl Clone for IpcServer {
23    fn clone(&self) -> Self {
24        Self {
25            socket_path: self.socket_path.clone(),
26            agent: self.agent.clone(),
27            tools: self.tools.clone(),
28            plugins: self.plugins.clone(),
29            session: self.session.clone(),
30        }
31    }
32}
33
34impl IpcServer {
35    pub fn new(socket_path: impl Into<String>) -> Self {
36        Self {
37            socket_path: socket_path.into(),
38            agent: Arc::new(AsyncMutex::new(Agent::new())),
39            tools: Arc::new(Mutex::new(ToolRegistry::new())),
40            plugins: Arc::new(Mutex::new(PluginRegistry::new())),
41            session: Arc::new(Mutex::new(Session::new("default", "default"))),
42        }
43    }
44
45    pub fn attach_agent(&self, agent: Agent) {
46        let agent_arc = self.agent.clone();
47        tokio::spawn(async move {
48            *agent_arc.lock().await = agent;
49        });
50    }
51
52    pub fn attach_tools(&self, tools: ToolRegistry) {
53        *self.tools.lock().unwrap() = tools;
54    }
55
56    pub fn attach_plugins(&self, plugins: PluginRegistry) {
57        *self.plugins.lock().unwrap() = plugins;
58    }
59
60    pub fn attach_session(&self, session: Session) {
61        *self.session.lock().unwrap() = session;
62    }
63
64    pub fn run(&self) -> std::io::Result<()> {
65        let path = Path::new(&self.socket_path);
66        if path.exists() {
67            std::fs::remove_file(path)?;
68        }
69        let listener = UnixListener::bind(path)?;
70        {
71            use std::os::unix::fs::PermissionsExt;
72            std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
73        }
74        info!("IPC server listening on {}", self.socket_path);
75
76        let runtime = tokio::runtime::Handle::try_current().map_err(std::io::Error::other)?;
77        let this = self.clone();
78        for stream in listener.incoming() {
79            match stream {
80                Ok(stream) => {
81                    let s = this.clone();
82                    runtime.spawn(async move {
83                        if let Err(e) = s.handle_connection(stream).await {
84                            warn!("connection error: {e}");
85                        }
86                    });
87                }
88                Err(e) => warn!("accept error: {e}"),
89            }
90        }
91        Ok(())
92    }
93
94    async fn handle_connection(
95        &self,
96        mut stream: std::os::unix::net::UnixStream,
97    ) -> std::io::Result<()> {
98        use std::io::{BufRead, BufReader, Write};
99        let reader = BufReader::new(stream.try_clone()?);
100        for line in reader.lines() {
101            let line = line?;
102            if line.is_empty() {
103                continue;
104            }
105            let response = self.handle_request(&line).await;
106            writeln!(stream, "{response}")?;
107        }
108        Ok(())
109    }
110
111    async fn handle_request(&self, line: &str) -> String {
112        let req: Value = match serde_json::from_str(line) {
113            Ok(v) => v,
114            Err(e) => return error_response(None, -32700, &format!("parse error: {e}")),
115        };
116        let id = req.get("id").cloned();
117        let method = req.get("method").and_then(|m| m.as_str()).unwrap_or("");
118        let params = req.get("params").cloned().unwrap_or(Value::Null);
119
120        let required_token = std::env::var("RX4_IPC_TOKEN")
121            .ok()
122            .filter(|s| !s.is_empty());
123        let provided = params.get("token").and_then(|t| t.as_str()).unwrap_or("");
124        let mutating = matches!(
125            method,
126            "prompt"
127                | "set_scope"
128                | "set_policy"
129                | "set_approver"
130                | "clear_authorizer"
131                | "cancel"
132                | "reset"
133                | "load_session"
134                | "save_session"
135        );
136        if method != "ping" {
137            match &required_token {
138                Some(token)
139                    if provided.len() != token.len()
140                        || !bool::from(provided.as_bytes().ct_eq(token.as_bytes())) =>
141                {
142                    return error_response(id, -32000, "invalid or missing token");
143                }
144                None if mutating => {
145                    return error_response(
146                        id,
147                        -32000,
148                        "RX4_IPC_TOKEN required for mutating IPC methods",
149                    );
150                }
151                _ => {}
152            }
153        }
154
155        let result: Result<Value, String> = match method {
156            "ping" => Ok(Value::String("pong".into())),
157            "state" => {
158                let agent = self.agent.lock().await;
159                Ok(serde_json::json!({
160                    "model": agent.model,
161                    "scope": agent.scope.name(),
162                    "policy_mode": format!("{:?}", agent.policy.mode),
163                    "shell_allow": agent.policy.shell_allow.len(),
164                    "shell_deny": agent.policy.shell_deny.len(),
165                    "has_approver": agent.approver.is_some(),
166                    "has_authorizer": agent.authorizer.is_some(),
167                    "tools": self.tools.lock().unwrap().count(),
168                    "plugins": self.plugins.lock().unwrap().count(),
169                }))
170            }
171            "tools" => Ok(Value::Array(self.tools.lock().unwrap().definitions())),
172            "plugins" => {
173                let p = self.plugins.lock().unwrap();
174                Ok(Value::Array(
175                    p.plugins
176                        .iter()
177                        .map(|pl| serde_json::json!({"id": pl.id, "name": pl.name}))
178                        .collect::<Vec<_>>(),
179                ))
180            }
181            "messages" => {
182                let agent = self.agent.lock().await;
183                let msgs = agent.messages.read();
184                Ok(Value::Array(
185                    msgs.iter()
186                        .map(|m| serde_json::json!({"role": m.role, "content": m.content}))
187                        .collect::<Vec<_>>(),
188                ))
189            }
190            "set_model" => {
191                let model = params
192                    .get("model")
193                    .and_then(|m| m.as_str())
194                    .unwrap_or("gpt-4o");
195                self.agent.lock().await.set_model(model);
196                Ok(Value::String(format!("model set to {model}")))
197            }
198            "set_scope" => {
199                if let Some(name) = params.get("scope").and_then(|s| s.as_str()) {
200                    if let Some(scope) = crate::mode::Scope::parse_scope(name) {
201                        self.agent.lock().await.set_scope(scope);
202                        Ok(Value::String(format!("scope set to {scope}")))
203                    } else {
204                        Err(format!("unknown scope: {name}"))
205                    }
206                } else {
207                    Err("missing scope".into())
208                }
209            }
210            "get_policy" => {
211                let agent = self.agent.lock().await;
212                serde_json::to_value(&agent.policy).map_err(|e| e.to_string())
213            }
214            "set_policy" => {
215                if let Some(raw) = params.get("policy").cloned() {
216                    match serde_json::from_value::<crate::permissions::Policy>(raw) {
217                        Ok(policy) => {
218                            self.agent.lock().await.set_policy(policy);
219                            Ok(Value::String("policy set".into()))
220                        }
221                        Err(e) => Err(e.to_string()),
222                    }
223                } else {
224                    Err("missing policy".into())
225                }
226            }
227            "set_approver" => {
228                // Host product Approver stays in-process; IPC only offers always_allow / always_deny.
229                let mode = params
230                    .get("mode")
231                    .and_then(|m| m.as_str())
232                    .unwrap_or("always_deny");
233                let mut agent = self.agent.lock().await;
234                match mode {
235                    "always_allow" | "allow" => {
236                        agent.set_approver(std::sync::Arc::new(crate::permissions::AlwaysAllow));
237                        Ok(Value::String(format!("approver set to {mode}")))
238                    }
239                    "always_deny" | "deny" => {
240                        agent.set_approver(std::sync::Arc::new(crate::permissions::AlwaysDeny));
241                        Ok(Value::String(format!("approver set to {mode}")))
242                    }
243                    "clear" | "none" => {
244                        agent.approver = None;
245                        Ok(Value::String("approver cleared".into()))
246                    }
247                    other => Err(format!("unknown approver mode: {other}")),
248                }
249            }
250            "clear_authorizer" => {
251                self.agent.lock().await.clear_authorizer();
252                Ok(Value::String("authorizer cleared".into()))
253            }
254            "prompt" => {
255                let text = params
256                    .get("text")
257                    .and_then(|t| t.as_str())
258                    .unwrap_or("")
259                    .to_string();
260                let agent = self.agent.clone();
261                tokio::spawn(async move {
262                    let mut a = agent.lock().await;
263                    let _ = a.prompt(&text).await;
264                });
265                Ok(Value::String("prompt accepted".into()))
266            }
267            "session_list" => {
268                let s = self.session.lock().unwrap();
269                Ok(serde_json::json!({"id": s.id, "entries": s.entries.len()}))
270            }
271            "session_clear" => {
272                self.session.lock().unwrap().entries.clear();
273                self.agent.lock().await.clear_messages();
274                Ok(Value::String("cleared".into()))
275            }
276            _ => return error_response(id, -32601, &format!("unknown method: {method}")),
277        };
278
279        match result {
280            Ok(value) => {
281                serde_json::json!({"jsonrpc": "2.0", "id": id, "result": value}).to_string()
282            }
283            Err(e) => error_response(id, -32603, &e),
284        }
285    }
286}
287
288fn error_response(id: Option<Value>, code: i32, message: &str) -> String {
289    serde_json::json!({
290        "jsonrpc": "2.0",
291        "id": id,
292        "error": {"code": code, "message": message}
293    })
294    .to_string()
295}
296
297#[cfg(test)]
298mod tests {
299    use super::*;
300    use crate::agent::Agent;
301
302    #[tokio::test]
303    async fn test_attach_agent() {
304        let server = IpcServer::new("/tmp/test_ipc_socket");
305        let mut new_agent = Agent::new();
306        new_agent.model = "test-model-abc".to_string();
307
308        server.attach_agent(new_agent);
309
310        // Yield to let the spawned task execute
311        tokio::task::yield_now().await;
312        // Just in case it needs a tiny bit of time
313        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
314
315        let agent_lock = server.agent.lock().await;
316        assert_eq!(agent_lock.model, "test-model-abc");
317    }
318}
rx4/ipc.rs

rx4/
ipc.rs
