monoloop_connector_codex/
lib.rs

1//! SPDX-License-Identifier: AGPL-3.0-or-later
2//! Copyright (C) Alexander R. Croft
3//!
4//! OpenAI Codex ACP connector profile.
5//!
6//! Speaks JSON-RPC 2.0 over **stdio NDJSON**, same client shape as Cursor / Agy ACP.
7//!
8//! **Practical path:** official adapter `@agentclientprotocol/codex-acp` (starts
9//! Codex App Server, maps ACP ↔ Codex). Configure via `CODEX_ACP_BIN` or default
10//! discovery (`codex-acp` → `npx --yes @agentclientprotocol/codex-acp`).
11//! Host auth: existing `codex login`, or `OPENAI_API_KEY` / `CODEX_API_KEY`.
12//!
13//! Session correlation identity is the ACP `sessionId` (explicit create/load only).
14//!
15//! The TypeScript/Python Codex SDK and `codex app-server` are alternate control
16//! planes; Monoloop folds Codex in via ACP so the shared Interpreter path applies.
17
18#![deny(missing_docs)]
19
20mod channel_binding;
21mod config;
22mod error;
23mod process;
24mod raw_dump;
25mod session;
26
27pub use channel_binding::{codex_channel_binding, CodexConnectorFactory};
28pub use config::{CodexAgentConfig, CodexSessionConfig};
29pub use error::CodexConnectorError;
30pub use raw_dump::CodexRawDump;
31pub use session::{CodexAgentHandle, CodexSession};
32
33use monoloop_connector::{
34    ConnectionCompletionHandle, ConnectionControlHandle, ConnectionEnd, ConnectionEndKind,
35    ConnectionOwnerWork, Connector, ConnectorDescriptor, ControlState, EndInitiator,
36    OpenConnection, OpenedRawConnection, PendingRawConnection, RawInputHandle, RawInputMessage,
37    RawOutputHandle,
38};
39use monoloop_contracts::{DialectBinding, DialectDescriptor, ExternalSessionId};
40use std::sync::Arc;
41use tokio::sync::{mpsc, oneshot};
42
43/// Factory for Codex ACP process connections.
44pub struct CodexConnector {
45    descriptor: ConnectorDescriptor,
46    default_config: CodexAgentConfig,
47}
48
49impl CodexConnector {
50    /// Create with default ACP bridge discovery.
51    pub fn new() -> Self {
52        Self {
53            descriptor: ConnectorDescriptor::codex_acp(),
54            default_config: CodexAgentConfig::default(),
55        }
56    }
57
58    /// Create with an explicit base config.
59    pub fn with_config(config: CodexAgentConfig) -> Self {
60        Self {
61            descriptor: ConnectorDescriptor::codex_acp(),
62            default_config: config,
63        }
64    }
65
66    /// High-level connect (preferred for multi-step sessions).
67    pub async fn connect(
68        &self,
69        config: CodexAgentConfig,
70    ) -> Result<CodexAgentHandle, CodexConnectorError> {
71        CodexAgentHandle::connect(config).await
72    }
73}
74
75impl Default for CodexConnector {
76    fn default() -> Self {
77        Self::new()
78    }
79}
80
81impl Connector for CodexConnector {
82    fn descriptor(&self) -> &ConnectorDescriptor {
83        &self.descriptor
84    }
85
86    /// `endpoint_ref`: `codex:stdio` / `stdio` or path to ACP bridge binary.
87    fn begin_open(&self, request: OpenConnection) -> PendingRawConnection {
88        let mut config = self.default_config.clone();
89        if let Some(path) = parse_endpoint(&request.endpoint_ref) {
90            config.command = path;
91            config.args.clear();
92        }
93        let connection_id = request.connection_id.clone();
94        let control_state = ControlState::new();
95        let control = ConnectionControlHandle::new(Arc::clone(&control_state));
96        let control_open = control.clone();
97        PendingRawConnection::open_owned(connection_id, control, async move {
98            open_raw(config, request, control_open, control_state).await
99        })
100    }
101}
102
103fn parse_endpoint(endpoint_ref: &str) -> Option<std::path::PathBuf> {
104    let s = endpoint_ref
105        .strip_prefix("codex:")
106        .or_else(|| endpoint_ref.strip_prefix("openai-codex:"))
107        .unwrap_or(endpoint_ref);
108    if s == "stdio" || s.is_empty() {
109        None
110    } else {
111        Some(std::path::PathBuf::from(s))
112    }
113}
114
115async fn open_raw(
116    config: CodexAgentConfig,
117    request: OpenConnection,
118    control: ConnectionControlHandle,
119    control_state: Arc<ControlState>,
120) -> Result<(OpenedRawConnection, ConnectionOwnerWork), monoloop_contracts::ConnectorError> {
121    let mut agent = CodexAgentHandle::connect(config.clone())
122        .await
123        .map_err(|e| e.into_connector_error())?;
124
125    let session = if let Some(ref ext) = request.external_session_id {
126        agent
127            .session_load(ext.as_str(), &config.cwd)
128            .await
129            .map_err(|e| e.into_connector_error())?
130    } else {
131        // D-026: CreationOnly MCP descriptor must reach provider session/new.
132        let mut session_cfg = CodexSessionConfig::new(&config.cwd);
133        if let Some(mcp) = request
134            .session_attachment
135            .as_ref()
136            .and_then(|a| a.initial_mcp.as_ref())
137        {
138            session_cfg.mcp_servers = serde_json::json!([{
139                "name": mcp.server_name,
140                "type": "http",
141                "url": mcp.expose_capability_url(),
142            }]);
143        }
144        agent
145            .session_new(session_cfg)
146            .await
147            .map_err(|e| e.into_connector_error())?
148    };
149
150    let dialect = DialectBinding::negotiated(DialectDescriptor::codex_acp("1"));
151    let max_chunk = request.limits.buffers.max_chunk_bytes.max(1);
152    let in_capacity = (request.limits.buffers.max_queued_input_bytes.max(1) / max_chunk).max(1);
153    let out_capacity = (request.limits.buffers.max_queued_output_bytes.max(1) / max_chunk)
154        .max(1)
155        .min(config.max_output_queue.max(1));
156    let (in_tx, mut in_rx) = mpsc::channel::<RawInputMessage>(in_capacity);
157    let (out_tx, out_rx) = mpsc::channel(out_capacity);
158    let (end_tx, end_rx) = oneshot::channel::<ConnectionEnd>();
159
160    let input = RawInputHandle::new(
161        request.connection_id.clone(),
162        in_tx,
163        Arc::clone(&control_state),
164        max_chunk,
165    );
166    let output = Arc::new(RawOutputHandle::new(
167        request.connection_id.clone(),
168        out_rx,
169        Arc::clone(&control_state),
170    ));
171    let completion = ConnectionCompletionHandle::new(end_rx);
172
173    let connection_id = request.connection_id.clone();
174    let external_session_id = Some(ExternalSessionId::new(session.session_id.clone()));
175
176    // Update pump must run concurrently with prompt_text awaits (LAW 23 joinable
177    // via JoinSet owned by this ConnectionOwnerWork — not fused into the input
178    // select, which deadlocks when prompt RPC waits on stdout).
179    let mut updates = agent.take_updates();
180    let control_wait = control.clone();
181    let owner_work = ConnectionOwnerWork::new(async move {
182        let mut joins = tokio::task::JoinSet::new();
183        let out_pump = out_tx;
184        joins.spawn(async move {
185            while let Some(bytes) = updates.recv().await {
186                if out_pump.send(bytes).await.is_err() {
187                    break;
188                }
189            }
190        });
191
192        let mut bytes_accepted = 0u64;
193        let (end_kind, initiated, safe_err) = loop {
194            tokio::select! {
195                biased;
196                _ = control_wait.interrupted() => {
197                    let kind = if control_state.terminate_requested() {
198                        ConnectionEndKind::Terminated
199                    } else {
200                        ConnectionEndKind::Cancelled
201                    };
202                    let _ = session.cancel().await;
203                    break (kind, EndInitiator::LocalControl, None);
204                }
205                msg = in_rx.recv() => {
206                    match msg {
207                        Some(RawInputMessage::Bytes(b)) => {
208                            bytes_accepted += b.len() as u64;
209                            let text = String::from_utf8_lossy(&b).into_owned();
210                            if text.trim().is_empty() {
211                                continue;
212                            }
213                            if let Err(e) = session.prompt_text(text).await {
214                                break (
215                                    ConnectionEndKind::TransportFailure,
216                                    EndInitiator::LocalTransport,
217                                    Some(safe_prompt_error(&e)),
218                                );
219                            }
220                        }
221                        Some(RawInputMessage::Finish) | None => {
222                            // Do not exit before shutdown joins the update pump —
223                            // Finish only ends input; updates may still be in flight.
224                            break (
225                                ConnectionEndKind::LocalShutdown,
226                                EndInitiator::LocalControl,
227                                None,
228                            );
229                        }
230                    }
231                }
232            }
233        };
234
235        agent.shutdown().await;
236        while joins.join_next().await.is_some() {}
237        control_state.mark_terminal();
238        let _ = end_tx.send(ConnectionEnd {
239            connection_id: connection_id.clone(),
240            kind: end_kind,
241            initiated_by: initiated,
242            bytes_accepted,
243            bytes_received: 0,
244            safe_transport_error: safe_err,
245        });
246    });
247
248    Ok((
249        OpenedRawConnection {
250            connection_id: request.connection_id,
251            external_session_id,
252            dialect,
253            input,
254            output,
255            control,
256            completion,
257        },
258        owner_work,
259    ))
260}
261
262fn safe_prompt_error(e: &CodexConnectorError) -> String {
263    match e.kind {
264        monoloop_contracts::ConnectorErrorKind::DeadlineExceeded => {
265            "prompt_rpc_deadline_exceeded".into()
266        }
267        monoloop_contracts::ConnectorErrorKind::Cancelled => "prompt_cancelled".into(),
268        monoloop_contracts::ConnectorErrorKind::ConnectionFailed => {
269            "prompt_connection_failed".into()
270        }
271        monoloop_contracts::ConnectorErrorKind::ProtocolFailed => "prompt_protocol_failed".into(),
272        monoloop_contracts::ConnectorErrorKind::SessionFailed => "prompt_session_failed".into(),
273        _ => "prompt_failed".into(),
274    }
275}
monoloop_connector_codex/lib.rs

monoloop_connector_codex/
lib.rs
