dscode_dap/
adapter.rs

1use crate::types::DebugSession;
2use serde_json::Value;
3use std::collections::HashMap;
4use std::process::Stdio;
5use std::sync::Arc;
6use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
7use tokio::process::{Child as TokioChild, Command as TokioCommand};
8use tokio::sync::{oneshot, Mutex};
9use tracing::{error, info, instrument};
10
11/// State machine for the DAP debug adapter lifecycle.
12///
13/// Tracks the lifecycle of a Debug Adapter Protocol (DAP) adapter process.
14///
15/// # State Diagram
16///
17/// ```text
18///   Stopped ──► Starting ──► Initializing ──► Configured ──► Running
19///     ▲             │              │               │            │
20///     │             │              │               │            │
21///     │             ▼              ▼               ▼            ▼
22///     │          Crashed ◄──── Crashed ◄──── Crashed ◄──── Crashed
23///     │                                                        │
24///     │                                                        │
25///     │                                            ShuttingDown│
26///     │                                                │       │
27///     └────────────────────────────────────────────────┘       │
28///                                                     ▲        │
29///                                                     └────────┘
30/// ```
31///
32/// # Transitions
33///
34/// - `Stopped` -> `Starting` (start() called)
35/// - `Starting` -> `Initializing` (process spawned, DAP initialize sent)
36/// - `Starting` -> `Crashed` (spawn failed)
37/// - `Initializing` -> `Configured` (initialize response received)
38/// - `Initializing` -> `Crashed` (initialize failed or timed out)
39/// - `Configured` -> `Running` (launch/attach response received)
40/// - `Configured` -> `Crashed` (launch/attach failed)
41/// - `Running` -> `ShuttingDown` (disconnect/terminate requested)
42/// - `Running` -> `Crashed` (adapter process exited unexpectedly)
43/// - `ShuttingDown` -> `Stopped` (adapter exited cleanly)
44/// - `Crashed` -> `Starting` (restart attempt)
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum DebugAdapterState {
47    /// No debug adapter process is running.
48    Stopped,
49    /// The debug adapter process is being spawned.
50    Starting,
51    /// The DAP initialize request has been sent and a response is pending.
52    Initializing,
53    /// The adapter has been initialized and configured, ready for launch/attach.
54    Configured,
55    /// The debug session is actively running with a live adapter.
56    Running,
57    /// A disconnect request has been sent and the adapter is shutting down.
58    ShuttingDown,
59    /// The adapter process exited unexpectedly or failed to start.
60    Crashed,
61}
62
63type PendingResponseMap = Arc<Mutex<HashMap<i32, oneshot::Sender<Result<Value, String>>>>>;
64
65/// A state-machine-based DAP client that manages a debug adapter process.
66///
67/// Handles spawning the debug adapter, performing the DAP initialize and
68/// launch/attach handshakes, sending requests, and processing responses.
69/// All state transitions are validated to maintain the lifecycle invariant.
70///
71/// # Concurrency
72///
73/// Same as [`LspClient`](crate::LspClient) -- state in `Arc<Mutex<>>`,
74/// separate from process/writer. Lock ordering: state -> process -> writer
75/// -> pending_responses.
76pub struct DebugAdapter {
77    state: Arc<Mutex<DebugAdapterState>>,
78    session: DebugSession,
79    process: Arc<Mutex<Option<TokioChild>>>,
80    writer: Arc<Mutex<Option<tokio::process::ChildStdin>>>,
81    pending_responses: PendingResponseMap,
82    adapter_command: String,
83    adapter_args: Vec<String>,
84    sequence: Arc<Mutex<i32>>,
85}
86
87impl std::fmt::Debug for DebugAdapter {
88    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89        f.debug_struct("DebugAdapter")
90            .field("session", &self.session)
91            .field("adapter_command", &self.adapter_command)
92            .field("adapter_args", &self.adapter_args)
93            .finish_non_exhaustive()
94    }
95}
96
97impl DebugAdapter {
98    /// Creates a new debug adapter in the `Stopped` state.
99    ///
100    /// - `session` — The debug session this adapter belongs to.
101    /// - `adapter_command` — The command to spawn the debug adapter (e.g., "/usr/bin/gdb").
102    /// - `adapter_args` — Arguments to pass to the adapter command.
103    pub fn new(session: DebugSession, adapter_command: String, adapter_args: Vec<String>) -> Self {
104        Self {
105            state: Arc::new(Mutex::new(DebugAdapterState::Stopped)),
106            session,
107            process: Arc::new(Mutex::new(None)),
108            writer: Arc::new(Mutex::new(None)),
109            pending_responses: Arc::new(Mutex::new(HashMap::new())),
110            adapter_command,
111            adapter_args,
112            sequence: Arc::new(Mutex::new(0)),
113        }
114    }
115
116    /// Validates and performs a state transition for the debug adapter.
117    pub(crate) async fn transition(&self, to: DebugAdapterState) -> Result<(), String> {
118        let mut state = self.state.lock().await;
119        let valid = match *state {
120            DebugAdapterState::Stopped => matches!(to, DebugAdapterState::Starting),
121            DebugAdapterState::Starting => {
122                matches!(to, DebugAdapterState::Initializing | DebugAdapterState::Crashed)
123            }
124            DebugAdapterState::Initializing => {
125                matches!(to, DebugAdapterState::Configured | DebugAdapterState::Crashed)
126            }
127            DebugAdapterState::Configured => {
128                matches!(to, DebugAdapterState::Running | DebugAdapterState::Crashed)
129            }
130            DebugAdapterState::Running => {
131                matches!(to, DebugAdapterState::ShuttingDown | DebugAdapterState::Crashed)
132            }
133            DebugAdapterState::ShuttingDown => matches!(to, DebugAdapterState::Stopped),
134            DebugAdapterState::Crashed => matches!(to, DebugAdapterState::Starting),
135        };
136
137        if valid {
138            info!(
139                id = %self.session.id,
140                from = ?*state,
141                to = ?to,
142                "State transition"
143            );
144            *state = to;
145            Ok(())
146        } else {
147            let msg = format!(
148                "Invalid state transition: {:?} -> {:?}",
149                *state, to
150            );
151            error!(
152                id = %self.session.id,
153                from = ?*state,
154                to = ?to,
155                "{}",
156                msg
157            );
158            Err(msg)
159        }
160    }
161
162    /// Returns the current state of the debug adapter.
163    pub async fn get_state(&self) -> DebugAdapterState {
164        *self.state.lock().await
165    }
166
167    /// Spawns the debug adapter process and begins the initialization sequence.
168    ///
169    /// Transitions from `Stopped` to `Starting` to `Initializing`. If spawning
170    /// fails, transitions to `Crashed`. A background task is spawned to read
171    /// responses from the adapter's stdout.
172    ///
173    /// Returns `Ok(())` if the process was spawned successfully, or an error
174    /// describing the failure.
175    #[instrument(skip(self))]
176    pub async fn start(&self) -> Result<(), String> {
177        self.transition(DebugAdapterState::Starting).await?;
178
179        let mut process_guard = self.process.lock().await;
180
181        if process_guard.is_some() {
182            return Ok(());
183        }
184
185        let mut cmd = TokioCommand::new(&self.adapter_command);
186        cmd.args(&self.adapter_args)
187            .stdin(Stdio::piped())
188            .stdout(Stdio::piped())
189            .stderr(Stdio::piped());
190
191        let mut child = match cmd.spawn() {
192            Ok(child) => child,
193            Err(e) => {
194                let _ = self.transition(DebugAdapterState::Crashed).await;
195                return Err(format!(
196                    "Failed to start debug adapter {}: {}",
197                    self.session.adapter_type, e
198                ));
199            }
200        };
201
202        let stdin = child.stdin.take().ok_or("Failed to get stdin")?;
203        let stdout = child.stdout.take().ok_or("Failed to get stdout")?;
204
205        if let Some(stderr) = child.stderr.take() {
206            let session_id = self.session.id.clone();
207            tokio::spawn(async move {
208                let reader = BufReader::new(stderr);
209                let mut lines = reader.lines();
210                while let Ok(Some(line)) = lines.next_line().await {
211                    error!(id = %session_id, "{}", line);
212                }
213            });
214        }
215
216        info!(
217            adapter_type = %self.session.adapter_type,
218            command = %self.adapter_command,
219            "Started debug adapter"
220        );
221
222        *self.writer.lock().await = Some(stdin);
223        *process_guard = Some(child);
224        // Release process lock before transitioning state (lock ordering: state -> process)
225        drop(process_guard);
226
227        let pending_clone = Arc::clone(&self.pending_responses);
228        let state_clone = Arc::clone(&self.state);
229        let session_id_clone = self.session.id.clone();
230        tokio::spawn(async move {
231            let mut reader = BufReader::new(stdout);
232            let mut header_buf = String::new();
233
234            loop {
235                header_buf.clear();
236                let mut read_err = false;
237                loop {
238                    let mut byte = [0u8; 1];
239                    if reader.read_exact(&mut byte).await.is_err() {
240                        read_err = true;
241                        break;
242                    }
243                    header_buf.push(byte[0] as char);
244
245                    if header_buf.ends_with("\r\n\r\n") {
246                        break;
247                    }
248
249                    if header_buf.len() > 4096 {
250                        error!("Header too long, disconnecting");
251                        read_err = true;
252                        break;
253                    }
254                }
255
256                if read_err {
257                    break;
258                }
259
260                let mut content_length: usize = 0;
261                for line in header_buf.split("\r\n") {
262                    if let Some(len_str) = line.strip_prefix("Content-Length: ") {
263                        content_length = len_str.trim().parse().unwrap_or(0);
264                    }
265                }
266
267                if content_length == 0 {
268                    continue;
269                }
270
271                let mut body = vec![0u8; content_length];
272                if reader.read_exact(&mut body).await.is_err() {
273                    break;
274                }
275
276                let response: Value = match serde_json::from_slice(&body) {
277                    Ok(v) => v,
278                    Err(e) => {
279                        error!("Failed to parse response: {}", e);
280                        continue;
281                    }
282                };
283
284                if response.get("type").and_then(|t| t.as_str()) == Some("response") {
285                    if let Some(seq) = response.get("request_seq").and_then(|v| v.as_i64()) {
286                        let mut pending_guard = pending_clone.lock().await;
287                        if let Some(sender) = pending_guard.remove(&(seq as i32)) {
288                            if response.get("success").and_then(|s| s.as_bool()) == Some(true) {
289                                let _ = sender.send(Ok(response));
290                            } else {
291                                let message = response
292                                    .get("message")
293                                    .and_then(|m| m.as_str())
294                                    .unwrap_or("Unknown error");
295                                let _ = sender.send(Err(message.to_string()));
296                            }
297                        }
298                    }
299                }
300            }
301
302            // Read loop ended -- transition to Crashed if still active
303            {
304                let mut state = state_clone.lock().await;
305                if matches!(
306                    *state,
307                    DebugAdapterState::Initializing
308                        | DebugAdapterState::Configured
309                        | DebugAdapterState::Running
310                ) {
311                    info!(
312                        id = %session_id_clone,
313                        "Read loop ended, state -> Crashed"
314                    );
315                    *state = DebugAdapterState::Crashed;
316                }
317            }
318            // Reject all pending requests when read loop ends
319            let mut pending = pending_clone.lock().await;
320            for (_, sender) in pending.drain() {
321                let _ = sender.send(Err("Debug adapter connection closed".to_string()));
322            }
323        });
324
325        self.transition(DebugAdapterState::Initializing).await?;
326
327        Ok(())
328    }
329
330    pub async fn stop(&self) -> Result<(), String> {
331        let _ = self.transition(DebugAdapterState::ShuttingDown).await;
332
333        let mut process_guard = self.process.lock().await;
334
335        if let Some(mut child) = process_guard.take() {
336            let _ = child.kill().await;
337            info!(id = %self.session.id, "Stopped debug adapter");
338        }
339
340        *self.writer.lock().await = None;
341
342        let _ = self.transition(DebugAdapterState::Stopped).await;
343
344        Ok(())
345    }
346
347    pub async fn is_running(&self) -> bool {
348        let state = self.state.lock().await;
349        matches!(
350            *state,
351            DebugAdapterState::Initializing
352                | DebugAdapterState::Configured
353                | DebugAdapterState::Running
354        )
355    }
356
357    async fn next_sequence(&self) -> i32 {
358        let mut seq = self.sequence.lock().await;
359        *seq += 1;
360        *seq
361    }
362
363    pub(crate) async fn send_request(
364        &self, command: &str, arguments: Option<Value>,
365    ) -> Result<Value, String> {
366        let seq = self.next_sequence().await;
367
368        let mut request = serde_json::json!({
369            "seq": seq,
370            "type": "request",
371            "command": command,
372        });
373
374        if let Some(args) = arguments {
375            request["arguments"] = args;
376        }
377
378        let (tx, rx) = oneshot::channel();
379        {
380            let mut pending = self.pending_responses.lock().await;
381            pending.insert(seq, tx);
382        }
383
384        {
385            let mut writer_guard = self.writer.lock().await;
386            let writer = writer_guard.as_mut().ok_or("Debug adapter stdin not available")?;
387
388            let body = serde_json::to_string(&request)
389                .map_err(|e| format!("Failed to serialize DAP request: {}", e))?;
390            let header = format!("Content-Length: {}\r\n\r\n", body.len());
391            writer
392                .write_all(header.as_bytes())
393                .await
394                .map_err(|e| format!("Failed to write DAP header: {}", e))?;
395            writer
396                .write_all(body.as_bytes())
397                .await
398                .map_err(|e| format!("Failed to write DAP body: {}", e))?;
399        }
400
401        match tokio::time::timeout(std::time::Duration::from_secs(30), rx).await {
402            Ok(Ok(Ok(response))) => Ok(response.get("body").cloned().unwrap_or(Value::Null)),
403            Ok(Ok(Err(e))) => Err(format!("DAP error: {}", e)),
404            Ok(Err(_)) => {
405                // Channel closed -- adapter crashed or connection lost
406                let mut pending = self.pending_responses.lock().await;
407                pending.remove(&seq);
408                Err("DAP request channel closed (adapter may have crashed)".to_string())
409            }
410            Err(_) => {
411                // Timeout -- adapter did not respond within 30 seconds
412                let mut pending = self.pending_responses.lock().await;
413                pending.remove(&seq);
414                Err("DAP request timed out after 30s".to_string())
415            }
416        }
417    }
418
419    #[allow(dead_code)]
420    pub(crate) async fn send_event(&self, event: &str, body: Option<Value>) -> Result<(), String> {
421        let seq = self.next_sequence().await;
422
423        let mut event_msg = serde_json::json!({
424            "seq": seq,
425            "type": "event",
426            "event": event,
427        });
428
429        if let Some(b) = body {
430            event_msg["body"] = b;
431        }
432
433        {
434            let mut writer_guard = self.writer.lock().await;
435            let writer = writer_guard.as_mut().ok_or("Debug adapter stdin not available")?;
436
437            let body = serde_json::to_string(&event_msg)
438                .map_err(|e| format!("Failed to serialize DAP event: {}", e))?;
439            let header = format!("Content-Length: {}\r\n\r\n", body.len());
440            writer
441                .write_all(header.as_bytes())
442                .await
443                .map_err(|e| format!("Failed to write DAP header: {}", e))?;
444            writer
445                .write_all(body.as_bytes())
446                .await
447                .map_err(|e| format!("Failed to write DAP body: {}", e))?;
448        }
449
450        Ok(())
451    }
452
453    #[instrument(skip(self))]
454    pub async fn initialize(&self) -> Result<Value, String> {
455        self.send_request(
456            "initialize",
457            Some(serde_json::json!({
458                "clientID": "dscode",
459                "clientName": "DSCode",
460                "adapterID": self.session.adapter_type,
461                "pathFormat": "path",
462                "linesStartAt1": true,
463                "columnsStartAt1": true,
464                "supportsVariableType": true,
465                "supportsVariablePaging": true,
466                "supportsRunInTerminalRequest": true,
467            })),
468        )
469        .await
470    }
471
472    pub async fn launch(&self, configuration: Value) -> Result<(), String> {
473        self.send_request("launch", Some(configuration)).await?;
474        Ok(())
475    }
476
477    pub async fn attach(&self, configuration: Value) -> Result<(), String> {
478        self.send_request("attach", Some(configuration)).await?;
479        Ok(())
480    }
481
482    pub async fn set_breakpoints(
483        &self, source: Value, breakpoints: Vec<crate::types::SourceBreakpoint>,
484    ) -> Result<Vec<crate::types::Breakpoint>, String> {
485        let response = self
486            .send_request(
487                "setBreakpoints",
488                Some(serde_json::json!({
489                    "source": source,
490                    "breakpoints": breakpoints,
491                })),
492            )
493            .await?;
494
495        let bps = response
496            .get("breakpoints")
497            .and_then(|b| b.as_array())
498            .ok_or("Invalid setBreakpoints response")?;
499
500        serde_json::from_value(Value::Array(bps.clone()))
501            .map_err(|e| format!("Failed to parse breakpoints: {}", e))
502    }
503
504    pub async fn continue_execution(&self, thread_id: i32) -> Result<(), String> {
505        self.send_request(
506            "continue",
507            Some(serde_json::json!({
508                "threadId": thread_id,
509            })),
510        )
511        .await?;
512        Ok(())
513    }
514
515    pub async fn pause(&self, thread_id: i32) -> Result<(), String> {
516        self.send_request(
517            "pause",
518            Some(serde_json::json!({
519                "threadId": thread_id,
520            })),
521        )
522        .await?;
523        Ok(())
524    }
525
526    pub async fn next(&self, thread_id: i32) -> Result<(), String> {
527        self.send_request(
528            "next",
529            Some(serde_json::json!({
530                "threadId": thread_id,
531            })),
532        )
533        .await?;
534        Ok(())
535    }
536
537    pub async fn step_in(&self, thread_id: i32) -> Result<(), String> {
538        self.send_request(
539            "stepIn",
540            Some(serde_json::json!({
541                "threadId": thread_id,
542            })),
543        )
544        .await?;
545        Ok(())
546    }
547
548    pub async fn step_out(&self, thread_id: i32) -> Result<(), String> {
549        self.send_request(
550            "stepOut",
551            Some(serde_json::json!({
552                "threadId": thread_id,
553            })),
554        )
555        .await?;
556        Ok(())
557    }
558
559    pub async fn disconnect(&self) -> Result<(), String> {
560        self.send_request("disconnect", None).await?;
561        Ok(())
562    }
563}
564
565impl Drop for DebugAdapter {
566    fn drop(&mut self) {}
567}
568
569#[cfg(test)]
570mod tests {
571    use super::*;
572    use crate::types::DebugState;
573
574    #[tokio::test]
575    async fn test_debug_adapter_state_transitions() {
576        let session = DebugSession {
577            id: "test-session".to_string(),
578            name: "Test Session".to_string(),
579            state: DebugState::Stopped,
580            adapter_type: "test-adapter".to_string(),
581        };
582        let adapter = DebugAdapter::new(
583            session,
584            "nonexistent-adapter".to_string(),
585            vec![],
586        );
587
588        // Initial state should be Stopped
589        assert_eq!(adapter.get_state().await, DebugAdapterState::Stopped);
590
591        // Valid: Stopped -> Starting
592        assert!(adapter.transition(DebugAdapterState::Starting).await.is_ok());
593        assert_eq!(adapter.get_state().await, DebugAdapterState::Starting);
594
595        // Valid: Starting -> Initializing
596        assert!(adapter.transition(DebugAdapterState::Initializing).await.is_ok());
597        assert_eq!(adapter.get_state().await, DebugAdapterState::Initializing);
598
599        // Valid: Initializing -> Configured
600        assert!(adapter.transition(DebugAdapterState::Configured).await.is_ok());
601        assert_eq!(adapter.get_state().await, DebugAdapterState::Configured);
602
603        // Valid: Configured -> Running
604        assert!(adapter.transition(DebugAdapterState::Running).await.is_ok());
605        assert_eq!(adapter.get_state().await, DebugAdapterState::Running);
606
607        // Valid: Running -> ShuttingDown
608        assert!(adapter.transition(DebugAdapterState::ShuttingDown).await.is_ok());
609        assert_eq!(adapter.get_state().await, DebugAdapterState::ShuttingDown);
610
611        // Valid: ShuttingDown -> Stopped
612        assert!(adapter.transition(DebugAdapterState::Stopped).await.is_ok());
613        assert_eq!(adapter.get_state().await, DebugAdapterState::Stopped);
614
615        // Test Crashed -> Starting restart path
616        // First go Stopped -> Starting -> Crashed
617        assert!(adapter.transition(DebugAdapterState::Starting).await.is_ok());
618        assert!(adapter.transition(DebugAdapterState::Crashed).await.is_ok());
619        assert_eq!(adapter.get_state().await, DebugAdapterState::Crashed);
620
621        // Valid: Crashed -> Starting (restart)
622        assert!(adapter.transition(DebugAdapterState::Starting).await.is_ok());
623
624        // Test invalid transitions
625        let adapter2 = DebugAdapter::new(
626            DebugSession {
627                id: "test-2".to_string(),
628                name: "Test 2".to_string(),
629                state: DebugState::Stopped,
630                adapter_type: "test".to_string(),
631            },
632            "noop".to_string(),
633            vec![],
634        );
635
636        // Invalid: Stopped -> Running (must go through Starting first)
637        assert!(adapter2.transition(DebugAdapterState::Running).await.is_err());
638        // Invalid: Stopped -> ShuttingDown
639        assert!(adapter2.transition(DebugAdapterState::ShuttingDown).await.is_err());
640    }
641}
dscode_dap/adapter.rs

dscode_dap/
adapter.rs
