agent_graph_mcp/
codex_app_server.rs

1//! Minimal, provider-neutral handling for Codex app-server JSON-RPC notifications.
2//!
3//! The transport emits numerous lifecycle/status events. Only `turn/completed`
4//! is terminal. Text streams in `item/agentMessage/delta`; older/newer servers
5//! may instead expose final text in a completed `agentMessage` item.
6
7use std::io::{BufRead, BufReader, Write};
8use std::os::unix::process::CommandExt;
9use std::path::Path;
10use std::process::{Command, Stdio};
11use std::sync::mpsc;
12use std::thread;
13use std::time::{Duration, Instant};
14
15use serde_json::{json, Value};
16
17/// Returns true only for the app-server's terminal turn notification.
18pub fn is_terminal_notification(event: &Value) -> bool {
19    event
20        .get("method")
21        .and_then(Value::as_str)
22        .is_some_and(|method| method == "turn/completed")
23}
24
25/// Collect assistant text without treating non-message events as output.
26///
27/// Delta text is authoritative when present. The completed-item form is a
28/// compatibility fallback for app-server versions that omit deltas.
29pub fn collect_text(events: &[Value]) -> String {
30    let deltas: String = events
31        .iter()
32        .filter(|event| {
33            event
34                .get("method")
35                .and_then(Value::as_str)
36                .is_some_and(|method| method == "item/agentMessage/delta")
37        })
38        .filter_map(|event| event.pointer("/params/delta").and_then(Value::as_str))
39        .collect();
40    if !deltas.is_empty() {
41        return deltas;
42    }
43    events
44        .iter()
45        .filter(|event| {
46            event
47                .get("method")
48                .and_then(Value::as_str)
49                .is_some_and(|method| method == "item/completed")
50        })
51        .filter(|event| {
52            event
53                .pointer("/params/item/type")
54                .and_then(Value::as_str)
55                .is_some_and(|kind| kind == "agentMessage")
56        })
57        .filter_map(|event| event.pointer("/params/item/text").and_then(Value::as_str))
58        .last()
59        .unwrap_or_default()
60        .to_owned()
61}
62
63fn remaining(deadline: Instant) -> Result<Duration, String> {
64    deadline
65        .checked_duration_since(Instant::now())
66        .filter(|duration| !duration.is_zero())
67        .ok_or_else(|| "codex app-server timed out".to_owned())
68}
69
70fn send(stdin: &mut impl Write, request: Value) -> Result<(), String> {
71    serde_json::to_writer(&mut *stdin, &request).map_err(|e| e.to_string())?;
72    stdin.write_all(b"\n").map_err(|e| e.to_string())?;
73    stdin.flush().map_err(|e| e.to_string())
74}
75
76fn receive(
77    rx: &mpsc::Receiver<Result<Value, String>>,
78    id: u64,
79    deadline: Instant,
80) -> Result<Value, String> {
81    loop {
82        let message = rx
83            .recv_timeout(remaining(deadline)?)
84            .map_err(|_| "codex app-server timed out waiting for response".to_owned())??;
85        if message.get("id").and_then(Value::as_u64) != Some(id) {
86            continue;
87        }
88        if let Some(error) = message.get("error") {
89            return Err(format!("codex app-server request {id} failed: {error}"));
90        }
91        return Ok(message.get("result").cloned().unwrap_or(Value::Null));
92    }
93}
94
95/// Run one read-only Codex app-server turn without exposing OAuth credentials.
96///
97/// The child owns its own Codex login/refresh lifecycle. Agent Graph passes no
98/// auth environment, token, or auth-store path. Server-initiated requests are
99/// denied because graph LLM nodes are prompt-only execution, not tool agents.
100pub fn run_turn(
101    codex_bin: &str,
102    model: &str,
103    cwd: &Path,
104    prompt: &str,
105    timeout: Duration,
106) -> Result<String, String> {
107    if model.trim().is_empty() {
108        return Err("codex model must not be empty".to_owned());
109    }
110    let mut child = unsafe {
111        Command::new(codex_bin)
112            .args([
113                "app-server",
114                "--stdio",
115                "-c",
116                &format!("model={:?}", model),
117                "-c",
118                "sandbox_mode=\"read-only\"",
119            ])
120            .current_dir(cwd)
121            .stdin(Stdio::piped())
122            .stdout(Stdio::piped())
123            .stderr(Stdio::null())
124            .pre_exec(|| {
125                // Create a new process group so we can cleanly kill the entire
126                // process tree on timeout/failure.  A negative PID in kill(2)
127                // targets the group.  setpgid(0,0) returns 0 on success.
128                if libc::setpgid(0, 0) == 0 {
129                    Ok(())
130                } else {
131                    Err(std::io::Error::last_os_error())
132                }
133            })
134            .spawn()
135    }
136    .map_err(|e| format!("failed to start codex app-server: {e}"))?;
137    let pgid = child.id();
138    let mut stdin = child.stdin.take().ok_or("codex stdin unavailable")?;
139    let stdout = child.stdout.take().ok_or("codex stdout unavailable")?;
140    let (tx, rx) = mpsc::channel();
141    thread::spawn(move || {
142        for line in BufReader::new(stdout).lines() {
143            let parsed = line
144                .map_err(|e| e.to_string())
145                .and_then(|line| serde_json::from_str::<Value>(&line).map_err(|e| e.to_string()));
146            if tx.send(parsed).is_err() {
147                return;
148            }
149        }
150    });
151    let deadline = Instant::now() + timeout;
152    // Startup gets a bounded sub-budget so a cold Node start doesn't
153    // consume the entire generation window.
154    let startup_deadline = deadline.min(Instant::now() + Duration::from_secs(60));
155    let result = (|| {
156        send(
157            &mut stdin,
158            json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"clientInfo":{"name":"agent-graph-mcp","title":"Agent Graph MCP","version":env!("CARGO_PKG_VERSION")},"capabilities":{}}}),
159        )?;
160        receive(&rx, 1, startup_deadline)?;
161        send(
162            &mut stdin,
163            json!({"jsonrpc":"2.0","method":"initialized","params":{}}),
164        )?;
165        send(
166            &mut stdin,
167            json!({"jsonrpc":"2.0","id":2,"method":"thread/start","params":{"cwd":cwd}}),
168        )?;
169        let thread = receive(&rx, 2, startup_deadline)?;
170        let thread_id = thread
171            .pointer("/thread/id")
172            .and_then(Value::as_str)
173            .or_else(|| thread.pointer("/thread/sessionId").and_then(Value::as_str))
174            .ok_or("codex thread/start returned no thread id")?;
175        send(
176            &mut stdin,
177            json!({"jsonrpc":"2.0","id":3,"method":"turn/start","params":{"threadId":thread_id,"input":[{"type":"text","text":prompt}]}}),
178        )?;
179        receive(&rx, 3, deadline)?;
180        let mut events = Vec::new();
181        loop {
182            let event = rx.recv_timeout(remaining(deadline)?).map_err(|_| {
183                "codex app-server timed out waiting for turn completion".to_owned()
184            })??;
185            if let Some(request_id) = event.get("id").and_then(Value::as_u64) {
186                send(
187                    &mut stdin,
188                    json!({"jsonrpc":"2.0","id":request_id,"error":{"code":-32000,"message":"Agent Graph forbids Codex tool/approval requests"}}),
189                )?;
190                continue;
191            }
192            let terminal = is_terminal_notification(&event);
193            events.push(event);
194            if terminal {
195                let text = collect_text(&events);
196                if text.trim().is_empty() {
197                    return Err("codex app-server completed without assistant text".to_owned());
198                }
199                return Ok(text);
200            }
201        }
202    })();
203    let _ = pgid_kill(pgid);
204    let _ = child.wait();
205    result
206}
207
208/// Send SIGKILL to every process in the process group identified by `pgid`
209/// (the codex app-server, its Node wrapper, and any subprocesses).
210fn pgid_kill(pgid: u32) {
211    // Graceful first: SIGTERM allows the Node wrapper to forward to children.
212    unsafe { libc::kill(-(pgid as i32), libc::SIGTERM) };
213    std::thread::sleep(std::time::Duration::from_millis(200));
214    // Force: SIGKILL guarantees termination of any stragglers.
215    unsafe { libc::kill(-(pgid as i32), libc::SIGKILL) };
216}
agent_graph_mcp/codex_app_server.rs

agent_graph_mcp/
codex_app_server.rs
