oxicode_agent/runtime/
dap.rs

1//! DAP (Debug Adapter Protocol) host — the reference [`DebugService`]
2//! implementation behind the `coding-omp-v1` "Debug service" extension.
3//!
4//! [`DapClient`] speaks the wire protocol (Content-Length framed JSON over
5//! the adapter's stdio): requests are matched to responses by `request_seq`,
6//! events accumulate in a shared log. [`DapDebugService`] manages a map of
7//! live sessions and implements the agent-side [`super::DebugService`]
8//! contract: `start` (initialize + launch/attach), `request` (typed DAP
9//! passthrough), `terminate` (terminate + disconnect + cleanup).
10//!
11//! Session config shape for `start`:
12//! `{ "adapter": ["debugpy", "--listen", ...], "request": "launch",
13//!    ...launchArguments }` — `adapter` is the adapter process command; all
14//! other keys are forwarded verbatim as the launch/attach arguments.
15
16use super::DebugService;
17use async_trait::async_trait;
18use parking_lot::Mutex;
19use std::collections::HashMap;
20use std::sync::Arc;
21use std::sync::atomic::{AtomicU64, Ordering};
22use std::time::{Duration, Instant};
23use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
24use tokio::process::{Child, ChildStdin, ChildStdout};
25
26const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
27
28/// A live DAP adapter connection.
29pub struct DapClient {
30    child: Child,
31    stdin: ChildStdin,
32    stdout: BufReader<ChildStdout>,
33    seq: AtomicU64,
34    /// Every adapter event received so far (bounded by the session's
35    /// lifetime; drained via [`DapClient::wait_for_event`]).
36    pub events: Arc<Mutex<Vec<serde_json::Value>>>,
37}
38
39impl std::fmt::Debug for DapClient {
40    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        f.debug_struct("DapClient")
42            .field("alive", &self.child.id().is_some())
43            .field("events", &self.events.lock().len())
44            .finish()
45    }
46}
47
48impl DapClient {
49    /// Spawn an adapter process and connect over its stdio.
50    pub fn spawn(adapter: &[String]) -> Result<Self, String> {
51        use tokio::process::Command;
52        let (program, args) = adapter.split_first().ok_or("adapter command is empty")?;
53        let mut cmd = Command::new(program);
54        cmd.args(args)
55            .stdin(std::process::Stdio::piped())
56            .stdout(std::process::Stdio::piped())
57            .stderr(std::process::Stdio::piped())
58            .kill_on_drop(true);
59        let mut child = cmd.spawn().map_err(|e| format!("spawn adapter: {e}"))?;
60        let stdin = child.stdin.take().ok_or("adapter: no stdin")?;
61        let stdout = child.stdout.take().ok_or("adapter: no stdout")?;
62        Ok(Self {
63            child,
64            stdin,
65            stdout: BufReader::new(stdout),
66            seq: AtomicU64::new(0),
67            events: Arc::new(Mutex::new(Vec::new())),
68        })
69    }
70
71    fn next_seq(&self) -> u64 {
72        self.seq.fetch_add(1, Ordering::SeqCst) + 1
73    }
74
75    async fn write_frame(&mut self, value: &serde_json::Value) -> Result<(), String> {
76        let body = serde_json::to_vec(value).map_err(|e| format!("serialize frame: {e}"))?;
77        self.stdin
78            .write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes())
79            .await
80            .map_err(|e| format!("adapter stdin write: {e}"))?;
81        self.stdin
82            .write_all(&body)
83            .await
84            .map_err(|e| format!("adapter stdin write: {e}"))?;
85        self.stdin
86            .flush()
87            .await
88            .map_err(|e| format!("adapter stdin flush: {e}"))
89    }
90
91    /// Read one wire frame with a deadline; returns the parsed JSON.
92    async fn read_frame(&mut self, deadline: Instant) -> Result<serde_json::Value, String> {
93        let mut content_length: Option<usize> = None;
94        loop {
95            let mut line = String::new();
96            let read = tokio::time::timeout_at(
97                tokio::time::Instant::from(deadline),
98                self.stdout.read_line(&mut line),
99            )
100            .await
101            .map_err(|_| "timed out reading adapter frame headers".to_string())?
102            .map_err(|e| format!("adapter stdout read: {e}"))?;
103            if read == 0 {
104                return Err("adapter closed the connection".to_string());
105            }
106            let trimmed = line.trim_end();
107            if trimmed.is_empty() {
108                break; // end of headers
109            }
110            if let Some(v) = trimmed.strip_prefix("Content-Length:") {
111                content_length = Some(
112                    v.trim()
113                        .parse()
114                        .map_err(|_| "bad Content-Length header".to_string())?,
115                );
116            }
117        }
118        let length = content_length.ok_or("adapter frame missing Content-Length")?;
119        let mut body = vec![0u8; length];
120        tokio::time::timeout_at(
121            tokio::time::Instant::from(deadline),
122            self.stdout.read_exact(&mut body),
123        )
124        .await
125        .map_err(|_| "timed out reading adapter frame body".to_string())?
126        .map_err(|e| format!("adapter stdout read: {e}"))?;
127        serde_json::from_slice(&body).map_err(|e| format!("bad adapter frame JSON: {e}"))
128    }
129
130    /// Issue a DAP request and await its response. Events seen along the way
131    /// are appended to [`Self::events`].
132    pub async fn request(
133        &mut self,
134        command: &str,
135        arguments: &serde_json::Value,
136    ) -> Result<serde_json::Value, String> {
137        let deadline = Instant::now() + REQUEST_TIMEOUT;
138        let seq = self.next_seq();
139        let frame = serde_json::json!({
140            "seq": seq,
141            "type": "request",
142            "command": command,
143            "arguments": arguments,
144        });
145        self.write_frame(&frame).await?;
146        loop {
147            if Instant::now() >= deadline {
148                return Err(format!("DAP request '{command}' timed out"));
149            }
150            let msg = self.read_frame(deadline).await?;
151            match msg.get("type").and_then(|t| t.as_str()) {
152                Some("response") => {
153                    let req_seq = msg.get("request_seq").and_then(|s| s.as_u64());
154                    if req_seq == Some(seq) {
155                        let success = msg
156                            .get("success")
157                            .and_then(|s| s.as_bool())
158                            .unwrap_or(false);
159                        return if success {
160                            Ok(msg.get("body").cloned().unwrap_or(serde_json::Value::Null))
161                        } else {
162                            Err(format!(
163                                "DAP '{command}' failed: {}",
164                                msg.get("message")
165                                    .and_then(|m| m.as_str())
166                                    .unwrap_or("unknown")
167                            ))
168                        };
169                    }
170                    // Response to someone else's request — keep scanning.
171                }
172                Some("event") => self.events.lock().push(msg),
173                _ => {}
174            }
175        }
176    }
177
178    /// Drain every buffered event of the given kind (best-effort wait).
179    pub async fn wait_for_event(
180        &mut self,
181        event: &str,
182        wait: Duration,
183    ) -> Option<serde_json::Value> {
184        let deadline = Instant::now() + wait;
185        loop {
186            {
187                let events = self.events.lock();
188                if let Some(found) = events
189                    .iter()
190                    .find(|e| e.get("event").and_then(|v| v.as_str()) == Some(event))
191                {
192                    return Some(found.clone());
193                }
194            }
195            if Instant::now() >= deadline {
196                return None;
197            }
198            match self.read_frame(deadline).await {
199                Ok(msg) => {
200                    if msg.get("type").and_then(|t| t.as_str()) == Some("event") {
201                        let is_match = msg.get("event").and_then(|v| v.as_str()) == Some(event);
202                        self.events.lock().push(msg);
203                        if is_match {
204                            return self
205                                .events
206                                .lock()
207                                .iter()
208                                .find(|e| e.get("event").and_then(|v| v.as_str()) == Some(event))
209                                .cloned();
210                        }
211                    }
212                }
213                Err(_) => return None,
214            }
215        }
216    }
217
218    async fn shutdown(&mut self) {
219        let _ = self
220            .request(
221                "disconnect",
222                &serde_json::json!({"terminateDebuggee": true}),
223            )
224            .await;
225        let _ = self.child.kill().await;
226    }
227}
228
229/// Reference [`DebugService`]: manages DAP adapter sessions.
230#[derive(Debug, Default)]
231pub struct DapDebugService {
232    sessions: Mutex<HashMap<String, DapClient>>,
233}
234
235impl DapDebugService {
236    /// Creates an empty service; adapter sessions are added by
237    /// [`DebugService::start`].
238    pub fn new() -> Self {
239        Self::default()
240    }
241}
242
243#[async_trait]
244impl DebugService for DapDebugService {
245    async fn start(&self, config: &serde_json::Value) -> Result<String, String> {
246        let adapter: Vec<String> = config
247            .get("adapter")
248            .and_then(|a| a.as_array())
249            .map(|items| {
250                items
251                    .iter()
252                    .filter_map(|v| v.as_str().map(String::from))
253                    .collect()
254            })
255            .ok_or("config must carry an \"adapter\" command array")?;
256        let request = config
257            .get("request")
258            .and_then(|r| r.as_str())
259            .unwrap_or("launch");
260        let mut launch_args = config.clone();
261        if let Some(obj) = launch_args.as_object_mut() {
262            obj.remove("adapter");
263            obj.insert(
264                "request".to_string(),
265                serde_json::Value::String(request.to_string()),
266            );
267        }
268
269        let mut client = DapClient::spawn(&adapter)?;
270        client
271            .request(
272                "initialize",
273                &serde_json::json!({"adapterID": config.get("type").and_then(|t| t.as_str()).unwrap_or("oxi"), "clientID": "oxicode"}),
274            )
275            .await?;
276        client.request(request, &launch_args).await?;
277        // Most adapters emit `stopped` right after launch/attach; bounded,
278        // non-fatal wait so sessions are observable immediately.
279        let _ = client
280            .wait_for_event("stopped", Duration::from_secs(10))
281            .await;
282
283        let session = uuid::Uuid::new_v4().to_string();
284        self.sessions.lock().insert(session.clone(), client);
285        Ok(session)
286    }
287
288    async fn request(
289        &self,
290        session: &str,
291        command: &str,
292        args: &serde_json::Value,
293    ) -> Result<serde_json::Value, String> {
294        // Take the session out so the lock is never held across `.await`
295        // (and one session's I/O never blocks another's).
296        let mut client = self
297            .sessions
298            .lock()
299            .remove(session)
300            .ok_or("unknown debug session")?;
301        let result = client.request(command, args).await;
302        self.sessions.lock().insert(session.to_string(), client);
303        result
304    }
305
306    async fn terminate(&self, session: &str) -> Result<(), String> {
307        let mut client = self
308            .sessions
309            .lock()
310            .remove(session)
311            .ok_or("unknown debug session")?;
312        let _ = client.request("terminate", &serde_json::json!({})).await;
313        client.shutdown().await;
314        Ok(())
315    }
316}
oxicode_agent/runtime/dap.rs

oxicode_agent/runtime/
dap.rs
