monoloop_connector_codex/
session.rs

1//! Codex ACP session lifecycle on one agent process.
2
3use crate::config::{CodexAgentConfig, CodexSessionConfig};
4use crate::error::CodexConnectorError;
5use crate::process::ProcessInner;
6use crate::raw_dump::CodexRawDump;
7use monoloop_contracts::{DialectBinding, DialectDescriptor, ExternalSessionId};
8use std::sync::Arc;
9use tokio::sync::mpsc;
10
11/// Live Codex ACP process + optional session.
12pub struct CodexAgentHandle {
13    inner: Arc<ProcessInner>,
14    updates: Option<mpsc::Receiver<bytes::Bytes>>,
15    dump: Arc<CodexRawDump>,
16}
17
18impl CodexAgentHandle {
19    /// Spawn ACP server (`codex-acp` / native), run initialize (+ optional authenticate).
20    pub async fn connect(config: CodexAgentConfig) -> Result<Self, CodexConnectorError> {
21        let dump = Arc::new(CodexRawDump::new(config.raw_dump_path.clone(), 10_000));
22        let (update_tx, updates) = mpsc::channel(config.max_output_queue);
23        let inner = ProcessInner::spawn(config.clone(), update_tx, Arc::clone(&dump)).await?;
24
25        let fs = if config.advertise_fs {
26            serde_json::json!({ "readTextFile": true, "writeTextFile": true })
27        } else {
28            serde_json::json!({ "readTextFile": false, "writeTextFile": false })
29        };
30
31        inner
32            .request(
33                "initialize",
34                serde_json::json!({
35                    "protocolVersion": 1,
36                    "clientCapabilities": {
37                        "fs": fs,
38                        "terminal": false
39                    },
40                    "clientInfo": {
41                        "name": config.client_name,
42                        "version": config.client_version
43                    }
44                }),
45            )
46            .await?;
47
48        if config.authenticate {
49            let _ = inner
50                .request(
51                    "authenticate",
52                    serde_json::json!({ "methodId": config.auth_method_id }),
53                )
54                .await?;
55        }
56
57        Ok(Self {
58            inner,
59            updates: Some(updates),
60            dump,
61        })
62    }
63
64    /// Take the session/update stream (once).
65    pub fn take_updates(&mut self) -> mpsc::Receiver<bytes::Bytes> {
66        self.updates
67            .take()
68            .expect("CodexAgentHandle updates already taken")
69    }
70
71    /// Create a new session (`session/new`).
72    pub async fn session_new(
73        &self,
74        config: CodexSessionConfig,
75    ) -> Result<CodexSession, CodexConnectorError> {
76        let result = self
77            .inner
78            .request(
79                "session/new",
80                serde_json::json!({
81                    "cwd": config.cwd.to_string_lossy(),
82                    "mcpServers": config.mcp_servers,
83                }),
84            )
85            .await?;
86        let session_id = result
87            .get("sessionId")
88            .and_then(|s| s.as_str())
89            .ok_or_else(|| CodexConnectorError::session("session/new missing sessionId"))?
90            .to_string();
91        let session = CodexSession {
92            session_id,
93            inner: Arc::clone(&self.inner),
94        };
95        if let Some(mode) = &config.mode_id {
96            // Best-effort: some bridges support set_mode; ignore protocol errors.
97            let _ = session.set_mode(mode).await;
98        }
99        Ok(session)
100    }
101
102    /// Explicit session load (no most-recent heuristic).
103    pub async fn session_load(
104        &self,
105        session_id: impl Into<String>,
106        cwd: impl AsRef<std::path::Path>,
107    ) -> Result<CodexSession, CodexConnectorError> {
108        let session_id = session_id.into();
109        let result = self
110            .inner
111            .request(
112                "session/load",
113                serde_json::json!({
114                    "sessionId": session_id,
115                    "cwd": cwd.as_ref().to_string_lossy(),
116                    "mcpServers": [],
117                }),
118            )
119            .await?;
120        let sid = result
121            .get("sessionId")
122            .and_then(|s| s.as_str())
123            .unwrap_or(&session_id)
124            .to_string();
125        Ok(CodexSession {
126            session_id: sid,
127            inner: Arc::clone(&self.inner),
128        })
129    }
130
131    /// Dialect binding for Interpreter.
132    pub fn dialect(&self) -> DialectBinding {
133        DialectBinding::negotiated(DialectDescriptor::codex_acp("1"))
134    }
135
136    /// Raw dump snapshot text.
137    pub fn raw_dump_text(&self) -> String {
138        self.dump.as_text()
139    }
140
141    /// Shared dump handle.
142    pub fn raw_dump(&self) -> Arc<CodexRawDump> {
143        Arc::clone(&self.dump)
144    }
145
146    /// Shut down the ACP process.
147    pub async fn shutdown(self) {
148        self.inner.shutdown().await;
149    }
150}
151
152/// One Codex session on an ACP process.
153pub struct CodexSession {
154    /// Authoritative session id.
155    pub session_id: String,
156    inner: Arc<ProcessInner>,
157}
158
159impl CodexSession {
160    /// Opaque external session id for Monoloop envelopes.
161    pub fn external_session_id(&self) -> ExternalSessionId {
162        ExternalSessionId::new(self.session_id.clone())
163    }
164
165    /// Set session mode when supported (`read-only` | `agent` | `agent-full-access`).
166    pub async fn set_mode(&self, mode_id: impl AsRef<str>) -> Result<(), CodexConnectorError> {
167        self.inner
168            .request(
169                "session/set_mode",
170                serde_json::json!({
171                    "sessionId": self.session_id,
172                    "modeId": mode_id.as_ref(),
173                }),
174            )
175            .await?;
176        Ok(())
177    }
178
179    /// Send `session/prompt` and wait for terminal RPC result.
180    pub async fn prompt_text(
181        &self,
182        text: impl Into<String>,
183    ) -> Result<serde_json::Value, CodexConnectorError> {
184        let text = text.into();
185        self.inner
186            .request(
187                "session/prompt",
188                serde_json::json!({
189                    "sessionId": self.session_id,
190                    "prompt": [{ "type": "text", "text": text }]
191                }),
192            )
193            .await
194    }
195
196    /// Cooperative cancel for this session.
197    pub async fn cancel(&self) -> Result<(), CodexConnectorError> {
198        let _ = self
199            .inner
200            .request(
201                "session/cancel",
202                serde_json::json!({ "sessionId": self.session_id }),
203            )
204            .await;
205        Ok(())
206    }
207}
monoloop_connector_codex/session.rs

monoloop_connector_codex/
session.rs
