mail4agent_messenger_shell/provider/
codex.rs

1//! Codex CLI: start a turn in a running thread through `codex app-server`.
2//!
3//! The operator runs one app-server (`codex app-server --listen unix://` or
4//! `--listen ws://127.0.0.1:PORT`) and attaches the interactive TUI to it
5//! with `codex --remote <same endpoint>`. This adapter is a second client on
6//! that server. Wire (verified against codex-cli 0.160.1 on the box):
7//! one JSON-RPC message per WebSocket text frame, no `"jsonrpc"` member;
8//! `initialize` → `initialized` → `turn/start {threadId, input}`. A thread
9//! the server has not loaded answers `thread not found`; the adapter then
10//! sends `thread/resume {threadId}` and retries `turn/start` once.
11//! `turn/start` on a thread with an active turn is accepted by the server.
12//! The other client (the TUI) sees `turn/started` and the user item.
13//!
14//! `unix://` uses a Unix domain socket and is unix-only here; Windows uses
15//! `ws://127.0.0.1:PORT`. Non-loopback `ws://` is refused.
16
17use std::io::{Read, Write};
18use std::path::PathBuf;
19use std::time::{Duration, Instant};
20
21use serde_json::{json, Value};
22use tungstenite::{Message, WebSocket};
23
24use super::{
25    wake_prompt, ProviderKind, ProviderSession, SessionKind, WakeAdapter, WakeError, WakeLetter,
26    WakeOutcome,
27};
28
29/// Env var naming the app-server endpoint (`unix:///path` or `ws://127.0.0.1:PORT`).
30pub const CODEX_APP_SERVER_ENV: &str = "M4A_CODEX_APP_SERVER";
31
32const RPC_TIMEOUT: Duration = Duration::from_secs(20);
33
34/// Where the shared app-server listens.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub enum CodexEndpoint {
37    /// `unix:///abs/path.sock` (WebSocket over a Unix socket).
38    Unix(PathBuf),
39    /// `ws://127.0.0.1:PORT` (loopback only).
40    Ws(String),
41}
42
43impl CodexEndpoint {
44    /// Parses `unix:///abs/path` or a loopback `ws://` URL. A bare
45    /// `unix://` (Codex default socket) is not resolved here; pass the path.
46    pub fn parse(value: &str) -> Option<Self> {
47        let value = value.trim();
48        if let Some(path) = value.strip_prefix("unix://") {
49            return (!path.is_empty()).then(|| CodexEndpoint::Unix(PathBuf::from(path)));
50        }
51        let rest = value.strip_prefix("ws://")?;
52        let host = rest.split(['/', '?']).next().unwrap_or("");
53        let host = host.rsplit_once(':').map(|(h, _)| h).unwrap_or(host);
54        matches!(host, "127.0.0.1" | "localhost" | "[::1]")
55            .then(|| CodexEndpoint::Ws(value.to_string()))
56    }
57
58    /// Reads [`CODEX_APP_SERVER_ENV`].
59    pub fn from_env() -> Option<Self> {
60        std::env::var(CODEX_APP_SERVER_ENV)
61            .ok()
62            .and_then(|value| Self::parse(&value))
63    }
64}
65
66/// Second client on the operator's Codex app-server.
67pub struct CodexAppServerAdapter {
68    endpoint: Option<CodexEndpoint>,
69}
70
71impl CodexAppServerAdapter {
72    /// Adapter for `endpoint` (`None`: every wake is `Unavailable`).
73    pub fn new(endpoint: Option<CodexEndpoint>) -> Self {
74        Self { endpoint }
75    }
76}
77
78impl WakeAdapter for CodexAppServerAdapter {
79    fn kind(&self) -> SessionKind {
80        SessionKind::local(ProviderKind::Codex)
81    }
82
83    fn probe(&self, _session: &ProviderSession) -> Result<(), WakeError> {
84        match &self.endpoint {
85            None => Err(WakeError::Unavailable(
86                "codex app-server endpoint unset".into(),
87            )),
88            Some(CodexEndpoint::Unix(path)) if !path.exists() => Err(WakeError::Unavailable(
89                "codex app-server socket missing".into(),
90            )),
91            Some(_) => Ok(()),
92        }
93    }
94
95    fn wake(
96        &mut self,
97        session: &ProviderSession,
98        letter: &WakeLetter<'_>,
99    ) -> Result<WakeOutcome, WakeError> {
100        self.probe(session)?;
101        let text = wake_prompt(session, letter);
102        match self.endpoint.clone() {
103            Some(CodexEndpoint::Ws(url)) => {
104                let (mut ws, _) = tungstenite::connect(url.as_str())
105                    .map_err(|err| WakeError::Unavailable(format!("codex ws: {err}")))?;
106                start_turn(&mut ws, &session.session_id, &text)
107            }
108            #[cfg(unix)]
109            Some(CodexEndpoint::Unix(path)) => {
110                let stream = std::os::unix::net::UnixStream::connect(&path)
111                    .map_err(|err| WakeError::Unavailable(format!("codex unix: {}", err.kind())))?;
112                stream
113                    .set_read_timeout(Some(RPC_TIMEOUT))
114                    .map_err(|err| WakeError::Transport(err.to_string()))?;
115                let (mut ws, _) = tungstenite::client("ws://localhost/", stream)
116                    .map_err(|err| WakeError::Unavailable(format!("codex handshake: {err}")))?;
117                start_turn(&mut ws, &session.session_id, &text)
118            }
119            #[cfg(not(unix))]
120            Some(CodexEndpoint::Unix(_)) => Err(WakeError::Unavailable(
121                "unix:// app-server is unix-only here; use ws://127.0.0.1:PORT".into(),
122            )),
123            None => Err(WakeError::Unavailable(
124                "codex app-server endpoint unset".into(),
125            )),
126        }
127    }
128}
129
130/// `initialize` request (exact shape verified on 0.160.1).
131pub fn initialize_request(id: i64) -> Value {
132    json!({
133        "id": id,
134        "method": "initialize",
135        "params": {"clientInfo": {"name": "mail4agent", "version": env!("CARGO_PKG_VERSION")}},
136    })
137}
138
139/// `turn/start` request with one text input.
140pub fn turn_start_request(id: i64, thread_id: &str, text: &str) -> Value {
141    json!({
142        "id": id,
143        "method": "turn/start",
144        "params": {"threadId": thread_id, "input": [{"type": "text", "text": text}]},
145    })
146}
147
148/// `thread/resume` request.
149pub fn thread_resume_request(id: i64, thread_id: &str) -> Value {
150    json!({"id": id, "method": "thread/resume", "params": {"threadId": thread_id}})
151}
152
153/// Runs the handshake and one `turn/start` (with one resume retry) on an
154/// open WebSocket. Generic over the stream so tests use a fake server.
155pub fn start_turn<S: Read + Write>(
156    ws: &mut WebSocket<S>,
157    thread_id: &str,
158    text: &str,
159) -> Result<WakeOutcome, WakeError> {
160    call(ws, initialize_request(0))?
161        .map_err(|msg| WakeError::Transport(format!("initialize: {msg}")))?;
162    send(ws, json!({"method": "initialized"}))?;
163    match call(ws, turn_start_request(1, thread_id, text))? {
164        Ok(_) => Ok(WakeOutcome::Delivered),
165        Err(msg) if msg.contains("not found") || msg.contains("not loaded") => {
166            call(ws, thread_resume_request(2, thread_id))?
167                .map_err(|msg| WakeError::Transport(format!("thread/resume: {msg}")))?;
168            call(ws, turn_start_request(3, thread_id, text))?
169                .map(|_| WakeOutcome::Delivered)
170                .map_err(|msg| WakeError::Transport(format!("turn/start: {msg}")))
171        }
172        Err(msg) => Err(WakeError::Transport(format!("turn/start: {msg}"))),
173    }
174}
175
176fn send<S: Read + Write>(ws: &mut WebSocket<S>, value: Value) -> Result<(), WakeError> {
177    ws.send(Message::text(value.to_string()))
178        .map_err(|err| WakeError::Transport(format!("codex send: {err}")))
179}
180
181/// Sends `request` and reads until the response with the same id.
182/// Notifications and server requests in between are skipped (approvals
183/// stay with the TUI). Inner `Err` is the server's error message.
184fn call<S: Read + Write>(
185    ws: &mut WebSocket<S>,
186    request: Value,
187) -> Result<Result<Value, String>, WakeError> {
188    let id = request["id"].clone();
189    send(ws, request)?;
190    let deadline = Instant::now() + RPC_TIMEOUT;
191    while Instant::now() < deadline {
192        let message = ws
193            .read()
194            .map_err(|err| WakeError::Transport(format!("codex read: {err}")))?;
195        let Message::Text(text) = message else {
196            continue;
197        };
198        let Ok(value) = serde_json::from_str::<Value>(text.as_str()) else {
199            continue;
200        };
201        if value.get("method").is_some() || value.get("id") != Some(&id) {
202            continue;
203        }
204        if let Some(error) = value.get("error") {
205            let msg = error["message"].as_str().unwrap_or("error").to_string();
206            return Ok(Err(msg));
207        }
208        return Ok(Ok(value.get("result").cloned().unwrap_or(Value::Null)));
209    }
210    Err(WakeError::Transport("codex rpc timed out".into()))
211}
212
213#[cfg(all(test, unix))]
214mod tests {
215    use super::*;
216    use crate::provider::tests::{letter, session};
217    use std::os::unix::net::UnixListener;
218
219    fn fake_server(path: PathBuf, loaded: bool) -> std::thread::JoinHandle<Vec<String>> {
220        let listener = UnixListener::bind(&path).unwrap();
221        std::thread::spawn(move || {
222            let (stream, _) = listener.accept().unwrap();
223            let mut ws = tungstenite::accept(stream).unwrap();
224            let mut seen = Vec::new();
225            let mut loaded = loaded;
226            loop {
227                let Ok(Message::Text(text)) = ws.read() else {
228                    break;
229                };
230                let v: Value = serde_json::from_str(text.as_str()).unwrap();
231                let method = v["method"].as_str().unwrap_or("").to_string();
232                seen.push(method.clone());
233                assert!(v.get("jsonrpc").is_none());
234                let reply = match method.as_str() {
235                    "initialize" => json!({"id": v["id"], "result": {"userAgent": "fake"}}),
236                    "initialized" => continue,
237                    "thread/resume" => {
238                        loaded = true;
239                        json!({"id": v["id"], "result": {"thread": {"id": "t"}}})
240                    }
241                    "turn/start" if loaded => {
242                        assert_eq!(v["params"]["input"][0]["type"], "text");
243                        ws.send(Message::text(
244                            json!({"method": "turn/started", "params": {}}).to_string(),
245                        ))
246                        .unwrap();
247                        json!({"id": v["id"], "result": {"turn": {"id": "u", "status": "inProgress"}}})
248                    }
249                    "turn/start" => {
250                        json!({"id": v["id"], "error": {"code": -32600, "message": "thread not found: t"}})
251                    }
252                    _ => json!({"id": v["id"], "error": {"code": -32601, "message": "nope"}}),
253                };
254                ws.send(Message::text(reply.to_string())).unwrap();
255                if seen.iter().filter(|m| *m == "turn/start").count() >= 1
256                    && loaded
257                    && method == "turn/start"
258                {
259                    break;
260                }
261            }
262            seen
263        })
264    }
265
266    fn sock(tag: &str) -> PathBuf {
267        let path =
268            std::env::temp_dir().join(format!("m4a-codex-{tag}-{}.sock", std::process::id()));
269        let _ = std::fs::remove_file(&path);
270        path
271    }
272
273    #[test]
274    fn loaded_thread_gets_one_turn_start() {
275        let path = sock("loaded");
276        let server = fake_server(path.clone(), true);
277        let mut adapter = CodexAppServerAdapter::new(Some(CodexEndpoint::Unix(path.clone())));
278        let s = session(SessionKind::local(ProviderKind::Codex));
279        assert_eq!(
280            adapter.wake(&s, &letter("hi")).unwrap(),
281            WakeOutcome::Delivered
282        );
283        assert_eq!(
284            server.join().unwrap(),
285            ["initialize", "initialized", "turn/start"]
286        );
287        let _ = std::fs::remove_file(&path);
288    }
289
290    #[test]
291    fn unloaded_thread_is_resumed_then_started() {
292        let path = sock("resume");
293        let server = fake_server(path.clone(), false);
294        let mut adapter = CodexAppServerAdapter::new(Some(CodexEndpoint::Unix(path.clone())));
295        let s = session(SessionKind::local(ProviderKind::Codex));
296        assert_eq!(
297            adapter.wake(&s, &letter("hi")).unwrap(),
298            WakeOutcome::Delivered
299        );
300        assert_eq!(
301            server.join().unwrap(),
302            [
303                "initialize",
304                "initialized",
305                "turn/start",
306                "thread/resume",
307                "turn/start"
308            ]
309        );
310        let _ = std::fs::remove_file(&path);
311    }
312
313    #[test]
314    fn endpoint_parse_is_loopback_only() {
315        assert_eq!(
316            CodexEndpoint::parse("unix:///tmp/a.sock"),
317            Some(CodexEndpoint::Unix("/tmp/a.sock".into()))
318        );
319        assert!(CodexEndpoint::parse("ws://127.0.0.1:4500").is_some());
320        assert!(CodexEndpoint::parse("ws://10.0.0.5:4500").is_none());
321        assert!(CodexEndpoint::parse("unix://").is_none());
322    }
323
324    /// Live: `M4A_LIVE_CODEX_APP_SERVER=unix:///…` and `M4A_LIVE_CODEX_THREAD=<id>`.
325    #[test]
326    #[ignore]
327    fn live_turn_start() {
328        let endpoint = std::env::var("M4A_LIVE_CODEX_APP_SERVER").unwrap();
329        let thread = std::env::var("M4A_LIVE_CODEX_THREAD").unwrap();
330        let mut adapter = CodexAppServerAdapter::new(CodexEndpoint::parse(&endpoint));
331        let mut s = session(SessionKind::local(ProviderKind::Codex));
332        s.session_id = thread;
333        assert_eq!(
334            adapter.wake(&s, &letter("live probe")).unwrap(),
335            WakeOutcome::Delivered
336        );
337    }
338}
mail4agent_messenger_shell/provider/codex.rs

mail4agent_messenger_shell/provider/
codex.rs
