fno_agents/
codex_inject.rs

1//! `mail-inject --harness codex`: LIVE delivery into a running codex session
2//! over the codex app-server daemon socket (US8, node x-d899). The codex sibling
3//! of [`crate::mail_inject`]'s claude `control.sock` path. Python's send path
4//! (`_mail_inject_codex`) runs this as a subprocess and falls back to the durable
5//! bus ONLY when it reports not-delivered (live-inject-first).
6//!
7//! Transport = JSON-RPC text frames over a WebSocket over a Unix socket (mirrors
8//! [`crate::logs_client`]). Unlike the claude path, the `turn/start` RPC RESPONSE
9//! is itself the delivery confirmation (the daemon accepts and queues the turn
10//! synchronously), so there is no transcript growth-poll.
11//!
12//! # The daemon prerequisite (why this can be a no-op)
13//!
14//! A default `codex` TUI runs its app-server IN-PROCESS with no socket on disk.
15//! The socket exists ONLY when a codex app-server daemon is running
16//! (`codex remote-control start`, standalone install + ChatGPT login); TUIs
17//! launched afterward auto-attach to it. Absent that daemon, `deliver_via_codex_daemon`
18//! returns `"no-daemon"` and the caller writes the durable floor. e2e verification
19//! needs the user's daemon; the pure builders + `classify_turn_start_response`
20//! below are the correct-by-construction unit-tested core.
21//!
22//! Protocol map verified against `~/code/tools/codex/codex-rs/` (app-server-protocol
23//! rpc.rs / v2/turn.rs, app-server-client remote.rs initialize handshake).
24
25use std::collections::HashSet;
26use std::path::{Path, PathBuf};
27use std::time::Duration;
28
29use futures_util::{SinkExt, StreamExt};
30use tokio::net::UnixStream;
31use tokio_tungstenite::tungstenite::Message;
32
33/// Overall budget for connect + handshake + turn/start round-trip. The daemon
34/// responds promptly; this only bounds a wedged socket so the verb cannot hang.
35const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
36
37/// Frame-skip ceiling per read: notifications and unrelated ids are skipped, but
38/// a chatty (or silent) socket must not loop forever before the timeout fires.
39const MAX_FRAMES: usize = 64;
40
41const LOADED_PAGE_SIZE: u32 = 100;
42const MAX_LOADED_PAGES: usize = 64;
43
44#[derive(Debug, serde::Serialize, PartialEq, Eq)]
45pub struct LoadedThread {
46    pub session_id: String,
47    pub cwd: String,
48}
49
50/// The shared codex app-server control socket: `$CODEX_HOME/app-server-control/
51/// app-server-control.sock` (CODEX_HOME defaults to `~/.codex`). Absent unless a
52/// codex app-server daemon is running.
53pub fn codex_app_server_socket_path() -> PathBuf {
54    let home = std::env::var("CODEX_HOME")
55        .ok()
56        .filter(|s| !s.is_empty())
57        .unwrap_or_else(|| format!("{}/.codex", std::env::var("HOME").unwrap_or_default()));
58    PathBuf::from(home)
59        .join("app-server-control")
60        .join("app-server-control.sock")
61}
62
63/// The `initialize` request frame. Local socket needs no auth/pairing — just the
64/// handshake. `id` is the string `"init"` (matched on the response). Note the
65/// absence of a `"jsonrpc":"2.0"` field: the codex app-server carries bare
66/// JSON-RPC text frames.
67pub fn initialize_request_json() -> String {
68    serde_json::json!({
69        "id": "init",
70        "method": "initialize",
71        "params": {
72            "clientInfo": {"name": "fno-mail-inject", "version": "0.1.0"},
73            "capabilities": {"experimentalApi": true}
74        }
75    })
76    .to_string()
77}
78
79/// The `initialized` notification (no `id`) sent after the initialize response.
80pub fn initialized_notification_json() -> String {
81    serde_json::json!({"method": "initialized"}).to_string()
82}
83
84/// The `turn/start` request injecting `text` into `thread_id` as a text input
85/// item. `id` is `1` (matched on the response). `text` is injected verbatim —
86/// the `<fno_mail>` envelope is rendered caller-side (Python), so this is a dumb
87/// transport, mirroring [`crate::mail_inject`].
88pub fn turn_start_request_json(thread_id: &str, text: &str) -> String {
89    serde_json::json!({
90        "id": 1,
91        "method": "turn/start",
92        "params": {
93            "threadId": thread_id,
94            "input": [{"type": "text", "text": text}]
95        }
96    })
97    .to_string()
98}
99
100pub fn loaded_list_request_json(id: u64, cursor: Option<&str>) -> String {
101    serde_json::json!({
102        "id": id,
103        "method": "thread/loaded/list",
104        "params": {
105            "cursor": cursor,
106            "limit": LOADED_PAGE_SIZE,
107        }
108    })
109    .to_string()
110}
111
112pub fn thread_read_request_json(id: u64, thread_id: &str) -> String {
113    serde_json::json!({
114        "id": id,
115        "method": "thread/read",
116        "params": {
117            "threadId": thread_id,
118            "includeTurns": false,
119        }
120    })
121    .to_string()
122}
123
124pub fn parse_loaded_list_response(
125    raw: &str,
126) -> Result<(Vec<String>, Option<String>), &'static str> {
127    let v: serde_json::Value = serde_json::from_str(raw).map_err(|_| "rpc-error")?;
128    if v.get("error").is_some() {
129        return Err("rpc-error");
130    }
131    let result = v
132        .get("result")
133        .and_then(|r| r.as_object())
134        .ok_or("rpc-error")?;
135    let data = result
136        .get("data")
137        .and_then(|d| d.as_array())
138        .ok_or("rpc-error")?;
139    let mut ids = Vec::with_capacity(data.len());
140    for item in data {
141        let id = item.as_str().filter(|s| !s.is_empty()).ok_or("rpc-error")?;
142        ids.push(id.to_string());
143    }
144    let next_cursor = match result.get("nextCursor") {
145        None | Some(serde_json::Value::Null) => None,
146        Some(value) => Some(value.as_str().ok_or("rpc-error")?.to_string()),
147    };
148    let next_cursor = next_cursor.filter(|cursor| !cursor.is_empty());
149    Ok((ids, next_cursor))
150}
151
152pub fn parse_thread_read_cwd(raw: &str) -> Option<String> {
153    let v: serde_json::Value = serde_json::from_str(raw).ok()?;
154    v.pointer("/result/thread/cwd")
155        .and_then(|cwd| cwd.as_str())
156        .map(str::to_string)
157}
158
159/// Classify a `turn/start` response frame into delivered / not-delivered.
160///
161/// - `.result.turn.id` is a string -> `Ok(())` (turn accepted; DELIVERED).
162/// - `.error` whose message mentions "not found"/"thread" -> `Err("thread-not-loaded")`
163///   (the session is embedded / not attached to the daemon -> durable fallback).
164/// - anything else (other rpc error, unparseable) -> `Err("rpc-error")`.
165///
166/// The `Err` value IS the `mail-inject` JSON `reason` token.
167pub fn classify_turn_start_response(raw: &str) -> Result<(), &'static str> {
168    let v: serde_json::Value = match serde_json::from_str(raw) {
169        Ok(v) => v,
170        Err(_) => return Err("rpc-error"),
171    };
172    if v.get("result")
173        .and_then(|r| r.get("turn"))
174        .and_then(|t| t.get("id"))
175        .and_then(|id| id.as_str())
176        .is_some()
177    {
178        return Ok(());
179    }
180    if let Some(err) = v.get("error") {
181        let msg = err
182            .get("message")
183            .and_then(|m| m.as_str())
184            .unwrap_or("")
185            .to_lowercase();
186        if msg.contains("not found") || msg.contains("thread") {
187            return Err("thread-not-loaded");
188        }
189    }
190    Err("rpc-error")
191}
192
193/// Deliver `text` into codex `thread_id` over the app-server daemon socket.
194/// `Ok(())` == turn accepted (delivered); every `Err(reason)` is a clean
195/// not-delivered signal whose value is the `mail-inject` JSON `reason` token.
196/// Socket absent -> `Err("no-daemon")`; a wedged socket -> `Err("io-error")`
197/// after [`HANDSHAKE_TIMEOUT`].
198pub async fn deliver_via_codex_daemon(thread_id: &str, text: &str) -> Result<(), &'static str> {
199    let sock = codex_app_server_socket_path();
200    if !sock.exists() {
201        return Err("no-daemon");
202    }
203    match tokio::time::timeout(HANDSHAKE_TIMEOUT, inject(&sock, thread_id, text)).await {
204        Ok(r) => r,
205        Err(_) => Err("io-error"),
206    }
207}
208
209pub async fn discover_loaded_threads() -> Result<Vec<LoadedThread>, &'static str> {
210    let sock = codex_app_server_socket_path();
211    if !sock.exists() {
212        return Err("no-daemon");
213    }
214    match tokio::time::timeout(HANDSHAKE_TIMEOUT, discover(&sock)).await {
215        Ok(result) => result,
216        Err(_) => Err("io-error"),
217    }
218}
219
220pub async fn run_loaded_thread_discovery() -> i32 {
221    let output = match discover_loaded_threads().await {
222        Ok(threads) => serde_json::json!({"available": true, "threads": threads}),
223        Err(reason) => serde_json::json!({"available": false, "reason": reason}),
224    };
225    println!("{output}");
226    0
227}
228
229/// The connect + initialize handshake + `turn/start` round-trip. Split out so
230/// [`deliver_via_codex_daemon`] can wrap it in a total timeout.
231async fn inject(sock: &Path, thread_id: &str, text: &str) -> Result<(), &'static str> {
232    let conn = UnixStream::connect(sock).await.map_err(|_| "io-error")?;
233    let ws = match tokio_tungstenite::client_async("ws://localhost/rpc", conn).await {
234        Ok((ws, _resp)) => ws,
235        Err(_) => return Err("handshake-failed"),
236    };
237    let (mut sink, mut stream) = ws.split();
238
239    sink.send(Message::Text(initialize_request_json().into()))
240        .await
241        .map_err(|_| "io-error")?;
242    read_until_id(&mut stream, &serde_json::json!("init")).await?;
243
244    sink.send(Message::Text(initialized_notification_json().into()))
245        .await
246        .map_err(|_| "io-error")?;
247
248    sink.send(Message::Text(
249        turn_start_request_json(thread_id, text).into(),
250    ))
251    .await
252    .map_err(|_| "io-error")?;
253    let resp = read_until_id(&mut stream, &serde_json::json!(1)).await?;
254    classify_turn_start_response(&resp)
255}
256
257async fn discover(sock: &Path) -> Result<Vec<LoadedThread>, &'static str> {
258    let conn = UnixStream::connect(sock).await.map_err(|_| "io-error")?;
259    let ws = match tokio_tungstenite::client_async("ws://localhost/rpc", conn).await {
260        Ok((ws, _resp)) => ws,
261        Err(_) => return Err("handshake-failed"),
262    };
263    let (mut sink, mut stream) = ws.split();
264
265    sink.send(Message::Text(initialize_request_json().into()))
266        .await
267        .map_err(|_| "io-error")?;
268    read_until_id(&mut stream, &serde_json::json!("init")).await?;
269    sink.send(Message::Text(initialized_notification_json().into()))
270        .await
271        .map_err(|_| "io-error")?;
272
273    let mut ids = Vec::new();
274    let mut seen_ids = HashSet::new();
275    let mut seen_cursors = HashSet::new();
276    let mut cursor: Option<String> = None;
277    let mut request_id = 2_u64;
278    let mut complete = false;
279    for _ in 0..MAX_LOADED_PAGES {
280        sink.send(Message::Text(
281            loaded_list_request_json(request_id, cursor.as_deref()).into(),
282        ))
283        .await
284        .map_err(|_| "io-error")?;
285        let raw = read_until_id(&mut stream, &serde_json::json!(request_id)).await?;
286        let (page, next_cursor) = parse_loaded_list_response(&raw)?;
287        for id in page {
288            if seen_ids.insert(id.clone()) {
289                ids.push(id);
290            }
291        }
292        request_id += 1;
293        match next_cursor {
294            None => {
295                complete = true;
296                break;
297            }
298            Some(next) if seen_cursors.insert(next.clone()) => cursor = Some(next),
299            Some(_) => return Err("rpc-error"),
300        }
301    }
302    if !complete {
303        return Err("rpc-error");
304    }
305
306    let mut threads = Vec::with_capacity(ids.len());
307    for session_id in ids {
308        sink.send(Message::Text(
309            thread_read_request_json(request_id, &session_id).into(),
310        ))
311        .await
312        .map_err(|_| "io-error")?;
313        let cwd = match read_until_id(&mut stream, &serde_json::json!(request_id)).await {
314            Ok(raw) => parse_thread_read_cwd(&raw).unwrap_or_default(),
315            Err(_) => String::new(),
316        };
317        request_id += 1;
318        threads.push(LoadedThread { session_id, cwd });
319    }
320    Ok(threads)
321}
322
323/// Read Text frames until one whose `id` equals `want`, returning its raw text.
324/// Skips notifications (no `id`) and frames for other ids; ignores non-Text
325/// frames. Bounded by [`MAX_FRAMES`]; a read error / closed stream is `"io-error"`.
326async fn read_until_id<S>(stream: &mut S, want: &serde_json::Value) -> Result<String, &'static str>
327where
328    S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
329{
330    for _ in 0..MAX_FRAMES {
331        match stream.next().await {
332            Some(Ok(Message::Text(t))) => {
333                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&t) {
334                    if v.get("id") == Some(want) {
335                        return Ok(t.to_string());
336                    }
337                }
338            }
339            Some(Ok(_)) => {} // non-Text frame (ping/binary/close-less); skip
340            Some(Err(_)) | None => return Err("io-error"),
341        }
342    }
343    Err("io-error")
344}
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349    use tokio::net::UnixListener;
350    use tokio_tungstenite::{accept_async, WebSocketStream};
351
352    async fn accept_initialized(listener: UnixListener) -> WebSocketStream<UnixStream> {
353        let (stream, _) = listener.accept().await.unwrap();
354        let mut ws = accept_async(stream).await.unwrap();
355        let init = ws.next().await.unwrap().unwrap().into_text().unwrap();
356        let init: serde_json::Value = serde_json::from_str(&init).unwrap();
357        assert_eq!(init["method"], "initialize");
358        ws.send(Message::Text(r#"{"id":"init","result":{}}"#.into()))
359            .await
360            .unwrap();
361        let initialized = ws.next().await.unwrap().unwrap().into_text().unwrap();
362        let initialized: serde_json::Value = serde_json::from_str(&initialized).unwrap();
363        assert_eq!(initialized["method"], "initialized");
364        ws
365    }
366
367    async fn next_request(ws: &mut WebSocketStream<UnixStream>) -> serde_json::Value {
368        let raw = ws.next().await.unwrap().unwrap().into_text().unwrap();
369        serde_json::from_str(&raw).unwrap()
370    }
371
372    #[test]
373    fn initialize_request_is_bare_jsonrpc_with_string_id() {
374        let v: serde_json::Value = serde_json::from_str(&initialize_request_json()).unwrap();
375        assert_eq!(v["id"], "init");
376        assert_eq!(v["method"], "initialize");
377        assert_eq!(v["params"]["clientInfo"]["name"], "fno-mail-inject");
378        // The codex app-server carries bare JSON-RPC frames: no "jsonrpc" field.
379        assert!(v.get("jsonrpc").is_none());
380    }
381
382    #[test]
383    fn initialized_notification_has_no_id() {
384        let v: serde_json::Value = serde_json::from_str(&initialized_notification_json()).unwrap();
385        assert_eq!(v["method"], "initialized");
386        assert!(v.get("id").is_none());
387    }
388
389    #[test]
390    fn turn_start_carries_thread_id_and_text_item() {
391        let v: serde_json::Value =
392            serde_json::from_str(&turn_start_request_json("THREAD-9", "hello MARKER")).unwrap();
393        assert_eq!(v["id"], 1);
394        assert_eq!(v["method"], "turn/start");
395        assert_eq!(v["params"]["threadId"], "THREAD-9");
396        assert_eq!(v["params"]["input"][0]["type"], "text");
397        assert_eq!(v["params"]["input"][0]["text"], "hello MARKER");
398    }
399
400    #[test]
401    fn loaded_list_request_carries_cursor_and_limit() {
402        let v: serde_json::Value =
403            serde_json::from_str(&loaded_list_request_json(7, Some("cursor-1"))).unwrap();
404        assert_eq!(v["id"], 7);
405        assert_eq!(v["method"], "thread/loaded/list");
406        assert_eq!(v["params"]["cursor"], "cursor-1");
407        assert_eq!(v["params"]["limit"], LOADED_PAGE_SIZE);
408    }
409
410    #[test]
411    fn loaded_list_parser_distinguishes_empty_and_malformed() {
412        let empty = r#"{"id":2,"result":{"data":[],"nextCursor":null}}"#;
413        assert_eq!(parse_loaded_list_response(empty), Ok((vec![], None)));
414        let page = r#"{"id":2,"result":{"data":["a","b"],"nextCursor":"b"}}"#;
415        assert_eq!(
416            parse_loaded_list_response(page),
417            Ok((vec!["a".into(), "b".into()], Some("b".into())))
418        );
419        assert_eq!(
420            parse_loaded_list_response(r#"{"id":2,"result":{"data":[1]}}"#),
421            Err("rpc-error")
422        );
423    }
424
425    #[test]
426    fn loaded_list_parser_treats_empty_cursor_as_terminal() {
427        let raw = r#"{"id":2,"result":{"data":["a"],"nextCursor":""}}"#;
428        assert_eq!(
429            parse_loaded_list_response(raw),
430            Ok((vec!["a".to_string()], None))
431        );
432    }
433
434    #[test]
435    fn thread_read_builder_and_parser_use_metadata_only() {
436        let v: serde_json::Value =
437            serde_json::from_str(&thread_read_request_json(9, "thread-1")).unwrap();
438        assert_eq!(v["method"], "thread/read");
439        assert_eq!(v["params"]["threadId"], "thread-1");
440        assert_eq!(v["params"]["includeTurns"], false);
441        let raw = r#"{"id":9,"result":{"thread":{"id":"thread-1","cwd":"/repo"}}}"#;
442        assert_eq!(parse_thread_read_cwd(raw).as_deref(), Some("/repo"));
443        assert_eq!(
444            parse_thread_read_cwd(r#"{"id":9,"error":{"message":"gone"}}"#),
445            None
446        );
447    }
448
449    #[tokio::test]
450    async fn discovery_paginates_deduplicates_and_keeps_failed_metadata() {
451        let temp = tempfile::tempdir().unwrap();
452        let socket = temp.path().join("codex.sock");
453        let listener = UnixListener::bind(&socket).unwrap();
454        let server = tokio::spawn(async move {
455            let mut ws = accept_initialized(listener).await;
456
457            let first = next_request(&mut ws).await;
458            assert_eq!(first["params"]["cursor"], serde_json::Value::Null);
459            ws.send(Message::Text(
460                r#"{"id":2,"result":{"data":["thread-a"],"nextCursor":"thread-a"}}"#.into(),
461            ))
462            .await
463            .unwrap();
464
465            let second = next_request(&mut ws).await;
466            assert_eq!(second["params"]["cursor"], "thread-a");
467            ws.send(Message::Text(
468                r#"{"id":3,"result":{"data":["thread-a","thread-b"],"nextCursor":null}}"#.into(),
469            ))
470            .await
471            .unwrap();
472
473            let read_a = next_request(&mut ws).await;
474            assert_eq!(read_a["params"]["threadId"], "thread-a");
475            ws.send(Message::Text(
476                r#"{"id":4,"result":{"thread":{"cwd":"/repo/a"}}}"#.into(),
477            ))
478            .await
479            .unwrap();
480
481            let read_b = next_request(&mut ws).await;
482            assert_eq!(read_b["params"]["threadId"], "thread-b");
483            ws.send(Message::Text(
484                r#"{"id":5,"error":{"message":"metadata unavailable"}}"#.into(),
485            ))
486            .await
487            .unwrap();
488        });
489
490        let threads = discover(&socket).await.unwrap();
491        assert_eq!(
492            threads,
493            vec![
494                LoadedThread {
495                    session_id: "thread-a".into(),
496                    cwd: "/repo/a".into(),
497                },
498                LoadedThread {
499                    session_id: "thread-b".into(),
500                    cwd: String::new(),
501                },
502            ]
503        );
504        server.await.unwrap();
505    }
506
507    #[tokio::test]
508    async fn discovery_rejects_repeated_cursor_without_partial_results() {
509        let temp = tempfile::tempdir().unwrap();
510        let socket = temp.path().join("codex.sock");
511        let listener = UnixListener::bind(&socket).unwrap();
512        let server = tokio::spawn(async move {
513            let mut ws = accept_initialized(listener).await;
514            let _first = next_request(&mut ws).await;
515            ws.send(Message::Text(
516                r#"{"id":2,"result":{"data":["thread-a"],"nextCursor":"thread-a"}}"#.into(),
517            ))
518            .await
519            .unwrap();
520            let _second = next_request(&mut ws).await;
521            ws.send(Message::Text(
522                r#"{"id":3,"result":{"data":["thread-b"],"nextCursor":"thread-a"}}"#.into(),
523            ))
524            .await
525            .unwrap();
526        });
527
528        assert_eq!(discover(&socket).await, Err("rpc-error"));
529        server.await.unwrap();
530    }
531
532    #[tokio::test]
533    async fn discovery_distinguishes_successful_empty_daemon() {
534        let temp = tempfile::tempdir().unwrap();
535        let socket = temp.path().join("codex.sock");
536        let listener = UnixListener::bind(&socket).unwrap();
537        let server = tokio::spawn(async move {
538            let mut ws = accept_initialized(listener).await;
539            let _list = next_request(&mut ws).await;
540            ws.send(Message::Text(
541                r#"{"id":2,"result":{"data":[],"nextCursor":null}}"#.into(),
542            ))
543            .await
544            .unwrap();
545        });
546
547        assert_eq!(discover(&socket).await, Ok(vec![]));
548        server.await.unwrap();
549    }
550
551    #[test]
552    fn classify_delivered_on_result_turn_id() {
553        let raw = r#"{"id":1,"result":{"turn":{"id":"turn-abc","status":"inProgress"}}}"#;
554        assert_eq!(classify_turn_start_response(raw), Ok(()));
555    }
556
557    #[test]
558    fn classify_thread_not_loaded_on_thread_error() {
559        let raw = r#"{"id":1,"error":{"code":-32000,"message":"thread not found"}}"#;
560        assert_eq!(classify_turn_start_response(raw), Err("thread-not-loaded"));
561    }
562
563    #[test]
564    fn classify_rpc_error_on_other_error_or_garbage() {
565        let other = r#"{"id":1,"error":{"code":-32601,"message":"method not implemented"}}"#;
566        assert_eq!(classify_turn_start_response(other), Err("rpc-error"));
567        assert_eq!(classify_turn_start_response("not json"), Err("rpc-error"));
568        // A result without a string turn id is not a confirmed delivery.
569        let no_turn = r#"{"id":1,"result":{}}"#;
570        assert_eq!(classify_turn_start_response(no_turn), Err("rpc-error"));
571    }
572
573    #[test]
574    fn socket_path_honors_codex_home() {
575        // Sanity: the tail is fixed; the head follows CODEX_HOME/HOME. We only
576        // assert the stable suffix to avoid mutating process env in a unit test.
577        let p = codex_app_server_socket_path();
578        assert!(p.ends_with("app-server-control/app-server-control.sock"));
579    }
580}
fno_agents/codex_inject.rs

fno_agents/
codex_inject.rs
