From 1e586d41a4394cc8f7d740bc45a55a11371db6d4 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Thu, 5 Nov 2026 10:21:25 +0100 Subject: [PATCH] feat!: host DAP directly in the image server Serve TCP debug sessions through the same image dispatcher as WebSocket clients. Connect VS Code directly to runa.dapAddress and remove the standalone adapter executable. BREAKING CHANGE: runa-dap is now library-only. Configure runa --dap-listen or RUNA_DAP_LISTEN and use runa.dapAddress instead of runa.debugAdapterPath. The server binds a loopback DAP listener on port 4711 by default. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 9 +- Cargo.lock | 4 +- README.md | 25 +- crates/runa-dap/Cargo.toml | 5 - crates/runa-dap/src/lib.rs | 35 +- crates/runa-dap/src/main.rs | 90 --- crates/runa-dap/tests/stdio.rs | 161 ------ crates/runa/Cargo.toml | 1 + crates/runa/src/dap.rs | 175 ++++++ crates/runa/src/lib.rs | 3 +- crates/runa/src/main.rs | 32 +- crates/runa/src/network.rs | 89 ++- .../adapter.rs => runa/tests/dap_adapter.rs} | 0 crates/runa/tests/dap_server.rs | 531 ++++++++++++++++++ crates/runa/tests/local_client.rs | 6 + docs/architecture.md | 9 +- docs/protocol.md | 26 +- mise.toml | 2 +- vscode/README.md | 33 +- vscode/integration/dap.test.ts | 52 +- vscode/integration/host.ts | 12 +- vscode/integration/server.ts | 14 +- vscode/package.json | 10 +- vscode/src/extension.ts | 13 +- 24 files changed, 907 insertions(+), 430 deletions(-) delete mode 100644 crates/runa-dap/src/main.rs delete mode 100644 crates/runa-dap/tests/stdio.rs create mode 100644 crates/runa/src/dap.rs rename crates/{runa-dap/tests/adapter.rs => runa/tests/dap_adapter.rs} (100%) create mode 100644 crates/runa/tests/dap_server.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4cc0c18..0ec2e4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,11 +11,9 @@ All notable changes to this project will be documented in this file. to opt out; Debug Actor focuses only the chosen actor in the shared session. - Add `mise run package-extension` to generate `vscode/runa.vsix` for manual installation without installing it automatically. -- Add `runa-dap --listen` / `RUNA_DAP_LISTEN` for loopback TCP DAP sessions, - retaining stdio as the default transport. -- Add `mise run install-extension` to build the debug adapter, package the VS Code +- Add `mise run install-extension` to package the VS Code extension, and install it locally using `code` (or `VSCODE_CLI`). -- Add a standalone Rust `runa-dap` stdio debug adapter for VS Code actor debugging. +- Add a Rust DAP library for VS Code actor debugging. - Expose `behavior.rn` for every actor in the graphical and VS Code trees, with actor-specific gutter breakpoints and source diagnostics. - Add an egui/eframe graphical client for browsing and inspecting actors in a @@ -30,6 +28,9 @@ All notable changes to this project will be documented in this file. ### Changed +- Host DAP directly in the image server at `127.0.0.1:4711`, configurable with + `--dap-listen` / `RUNA_DAP_LISTEN`. VS Code connects with `runa.dapAddress`; + remove the standalone `runa-dap` executable and `runa.debugAdapterPath`. - Replace the graphical client's behavior dropdowns with a left-hand source file tree and a right-hand actor tree with metadata, state, mailbox, log, and behavior entries. diff --git a/Cargo.lock b/Cargo.lock index 738771d..49ffd1f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2554,6 +2554,7 @@ dependencies = [ "runa-actor", "runa-client", "runa-code", + "runa-dap", "runa-image-api", "runa-protocol", "serde_json", @@ -2618,9 +2619,6 @@ dependencies = [ name = "runa-dap" version = "0.0.0" dependencies = [ - "clap", - "runa", - "runa-client", "runa-core", "runa-protocol", "serde", diff --git a/README.md b/README.md index b0af7aa..5318b9c 100644 --- a/README.md +++ b/README.md @@ -187,11 +187,14 @@ print results without a REPL banner, and exit unsuccessfully on errors. exits either interface without stopping the image. `cancel` is no longer a REPL command; the underlying runtime cancellation API remains available. -The server is WebSocket-only; stdio transport has been removed. By default it listens on +The image API uses WebSockets; stdio transport has been removed. By default it listens on `ws://127.0.0.1:3737/` and requires the `runa-debug` subprotocol, with up to 16 simultaneous clients sharing one live image. The listener rejects non-loopback addresses and has no authentication; any local process that can connect has full access to the development API. Use it only on a trusted local machine. +The same server hosts a DAP TCP endpoint at `127.0.0.1:4711`. Override it with +`--dap-listen` or `RUNA_DAP_LISTEN`; port `0` selects an available port reported +on stderr. Both endpoints are restricted to loopback and share the same image. The `runa-client` crate connects over loopback WebSockets and exchange typed protocol requests with request/response correlation. @@ -264,15 +267,15 @@ the runtime contract. ## VS Code client To install into your regular VS Code instance, run `npm --prefix vscode ci` -followed by `mise run install-extension`, then reload the window. This builds the -Rust debug adapter and packages/installs the extension using the local `code` +followed by `mise run install-extension`, then reload the window. This +packages/installs the extension using the local `code` CLI. Set `VSCODE_CLI` to choose another executable, such as `code-insiders`. See [extension setup](vscode/README.md#run-from-source) for debugger configuration. To generate an installable file without installing it, run `mise run package-extension`. The output is `vscode/runa.vsix`, ready for -VS Code's **Extensions: Install from VSIX...** command. The Rust debug adapter -is built/installed separately. +VS Code's **Extensions: Install from VSIX...** command. No separate debug adapter +executable is needed. With Node.js 22+ and npm installed, run `npm --prefix vscode ci`. In VS Code, open this repository and launch **Run Runa Extension** from Run and Debug @@ -305,12 +308,10 @@ temporary script). Saving compiles an actor-specific override and installs it at the next handler boundary without changing shared source, state, or mailbox. Every actor directory includes `behavior.rn`, even without a temporary override. Set breakpoints there and use **Debug Actor** to debug that actor with VS Code's -Run and Debug controls. Build the standalone Rust DAP server with -`cargo build -p runa-dap` and set `runa.debugAdapterPath` to its absolute path, -or install it on PATH with `cargo install --path crates/runa-dap`. -For TCP DAP clients, use `runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737` -(or set `RUNA_DAP_LISTEN`). The DAP listener is loopback-only; omitting the -listen option keeps stdio transport. +Run and Debug controls. The image server hosts DAP directly at `127.0.0.1:4711`; +set `runa.dapAddress` to match a custom server `--dap-listen` address. +The old standalone `runa-dap` executable and `runa.debugAdapterPath` setting +have been removed. **Reload Actor Source** clears the override, **Go to Behavior** opens the shared source, and **Update Behavior** asks before copying the temporary text into shared working source without committing. @@ -403,7 +404,7 @@ use `xvfb-run -a mise run test-extension-host`. ├── crates/ │ ├── runa/ # Image server, protocol dispatcher, WebSocket transport │ ├── runa-cli/ # Interactive REPL client -│ ├── runa-dap/ # Standalone stdio Debug Adapter Protocol server +│ ├── runa-dap/ # Debug Adapter Protocol library hosted by the image server │ ├── runa-core/ # Shared stable identities and portable actor values │ ├── runa-actor/ # Actor value contracts │ ├── runa-code/ # Source identities and compile cache diff --git a/crates/runa-dap/Cargo.toml b/crates/runa-dap/Cargo.toml index 41a2c05..b4b6af9 100644 --- a/crates/runa-dap/Cargo.toml +++ b/crates/runa-dap/Cargo.toml @@ -4,12 +4,7 @@ version = "0.0.0" edition = "2024" [dependencies] -clap.workspace = true -runa-client = { path = "../runa-client" } runa-core = { path = "../runa-core" } runa-protocol = { path = "../runa-protocol" } serde = { version = "1", features = ["derive"] } serde_json = "1" - -[dev-dependencies] -runa = { path = "../runa" } diff --git a/crates/runa-dap/src/lib.rs b/crates/runa-dap/src/lib.rs index 2145a90..77b1010 100644 --- a/crates/runa-dap/src/lib.rs +++ b/crates/runa-dap/src/lib.rs @@ -1,11 +1,10 @@ -//! One-actor DAP sessions over the loopback image protocol. Debug controls never tick the image. +//! Actor and image DAP sessions over a request [`Connection`] to an image. +//! The image server hosts these sessions; debug controls never tick the image. use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet}; -use std::net::SocketAddr; use std::rc::Rc; -use runa_client::WebSocketClient; use runa_protocol::{ ActorSnapshotDto, ActorStatusDto, Command, ExecutionInspectionDto, Reply, SourceBreakpoint, StackFrame, StepMode, SystemMessageDto, @@ -20,36 +19,6 @@ pub trait Connection { fn request(&mut self, command: Command) -> Result; } -pub struct WebSocketConnection { - address: SocketAddr, - client: Option, -} - -impl WebSocketConnection { - pub fn new(address: SocketAddr) -> Self { - Self { - address, - client: None, - } - } -} - -impl Connection for WebSocketConnection { - fn request(&mut self, command: Command) -> Result { - if self.client.is_none() { - self.client = - Some(WebSocketClient::connect(self.address).map_err(|error| error.to_string())?); - } - self.client - .as_mut() - .unwrap() - .request(command) - .map_err(|error| error.to_string())? - .result - .map_err(|error| error.message) - } -} - #[derive(Deserialize)] struct Request { seq: u64, diff --git a/crates/runa-dap/src/main.rs b/crates/runa-dap/src/main.rs deleted file mode 100644 index 52051ed..0000000 --- a/crates/runa-dap/src/main.rs +++ /dev/null @@ -1,90 +0,0 @@ -use std::io::{self, BufReader, Read, Write}; -use std::net::{Shutdown, SocketAddr, TcpListener}; -use std::sync::mpsc::{self, RecvTimeoutError}; -use std::time::Duration; - -use clap::Parser; -use runa_dap::{Adapter, WebSocketConnection, framing}; - -#[derive(Parser)] -#[command(about = "Stdio or loopback TCP Debug Adapter Protocol server for Runa actors")] -struct Args { - #[arg(long, env = "RUNA_CONNECT", default_value = "127.0.0.1:3737")] - connect: SocketAddr, - /// Listen for TCP DAP sessions instead of using stdin/stdout. - #[arg(long, env = "RUNA_DAP_LISTEN")] - listen: Option, -} - -fn main() -> io::Result<()> { - let args = Args::parse(); - if !args.connect.ip().is_loopback() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "--connect must be a loopback address", - )); - } - if let Some(address) = args.listen { - if !address.ip().is_loopback() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "--listen must be a loopback address", - )); - } - let listener = TcpListener::bind(address)?; - eprintln!("runa-dap: listening on {}", listener.local_addr()?); - for stream in listener.incoming() { - let stream = stream?; - let result = run_session(stream.try_clone()?, &stream, args.connect); - if let Err(error) = stream.shutdown(Shutdown::Both) { - eprintln!("runa-dap: could not close session socket: {error}"); - } - if let Err(error) = result { - eprintln!("runa-dap: session failed: {error}"); - } - } - Ok(()) - } else { - run_session(io::stdin(), io::stdout().lock(), args.connect) - } -} - -fn run_session( - input: impl Read + Send + 'static, - mut output: impl Write, - connect: SocketAddr, -) -> io::Result<()> { - let (sender, receiver) = mpsc::sync_channel(16); - std::thread::spawn(move || { - let mut input = BufReader::new(input); - loop { - let result = framing::read_frame(&mut input); - let finished = !matches!(&result, Ok(Some(_))); - if sender.send(result).is_err() || finished { - break; - } - } - }); - let mut adapter = Adapter::new(WebSocketConnection::new(connect)); - while !adapter.is_closed() { - let messages = match receiver.recv_timeout(Duration::from_millis(150)) { - Ok(Ok(Some(message))) => adapter.handle(message), - Ok(Ok(None)) | Err(RecvTimeoutError::Disconnected) => break, - Ok(Err(error)) => { - for message in adapter.terminate(&format!("Invalid DAP framing: {error}")) { - framing::write_frame(&mut output, &message)?; - } - return Err(error); - } - Err(RecvTimeoutError::Timeout) => adapter.poll(), - }; - for message in messages { - framing::write_frame(&mut output, &message)?; - } - // Poll even under a steady stream of input; requests cannot starve lifecycle checks. - for message in adapter.poll() { - framing::write_frame(&mut output, &message)?; - } - } - Ok(()) -} diff --git a/crates/runa-dap/tests/stdio.rs b/crates/runa-dap/tests/stdio.rs deleted file mode 100644 index 68dc4c4..0000000 --- a/crates/runa-dap/tests/stdio.rs +++ /dev/null @@ -1,161 +0,0 @@ -use std::io::{BufRead, BufReader, Cursor, Write}; -use std::net::TcpStream; -use std::process::{Command, Stdio}; -use std::time::Duration; - -use runa_dap::framing::{read_frame, write_frame}; -use serde_json::{Value, json}; - -fn run(input: &[u8], args: &[&str]) -> std::process::Output { - let mut child = Command::new(env!("CARGO_BIN_EXE_runa-dap")) - .args(args) - .env_remove("RUNA_CONNECT") - .env_remove("RUNA_DAP_LISTEN") - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(); - child.stdin.take().unwrap().write_all(input).unwrap(); - child.wait_with_output().unwrap() -} - -fn messages(output: &[u8]) -> Vec { - let mut reader = Cursor::new(output); - let mut messages = Vec::new(); - while let Some(message) = read_frame(&mut reader).unwrap() { - messages.push(message); - } - messages -} - -#[test] -fn executable_exchanges_standard_frames_and_disconnects_without_a_server() { - let mut input = Vec::new(); - write_frame( - &mut input, - &json!({"seq": 1, "type": "request", "command": "initialize"}), - ) - .unwrap(); - write_frame( - &mut input, - &json!({"seq": 2, "type": "request", "command": "launch"}), - ) - .unwrap(); - write_frame( - &mut input, - &json!({"seq": 3, "type": "request", "command": "disconnect"}), - ) - .unwrap(); - let output = run(&input, &[]); - assert!(output.status.success(), "{:?}", output.stderr); - let messages = messages(&output.stdout); - assert_eq!( - messages[0]["body"]["supportsConfigurationDoneRequest"], - true - ); - assert_eq!(messages[1]["success"], false); - assert_eq!(messages[2]["success"], true); - assert_eq!(messages[3]["event"], "terminated"); - for (index, message) in messages.iter().enumerate() { - assert_eq!(message["seq"], index + 1); - } -} - -#[test] -fn executable_reports_bad_framing_on_protocol_and_stderr() { - let output = run(b"Content-Length: 1048577\r\n\r\n", &[]); - assert!(!output.status.success()); - assert!(!output.stderr.is_empty()); - let messages = messages(&output.stdout); - assert_eq!(messages[0]["event"], "output"); - assert!( - messages[0]["body"]["output"] - .as_str() - .unwrap() - .contains("Invalid DAP framing") - ); - assert_eq!(messages[1]["event"], "terminated"); -} - -#[test] -fn cli_documents_connect_and_rejects_nonloopback_addresses() { - let help = run(&[], &["--help"]); - assert!(help.status.success()); - let help = String::from_utf8(help.stdout).unwrap(); - assert!(help.contains("RUNA_CONNECT")); - assert!(help.contains("127.0.0.1:3737")); - assert!(help.contains("--listen")); - assert!(help.contains("RUNA_DAP_LISTEN")); - for address in ["0.0.0.0:3737", "192.0.2.1:3737", "invalid"] { - let output = run(&[], &["--connect", address]); - assert!(!output.status.success()); - assert!(!output.stderr.is_empty()); - } - for address in ["0.0.0.0:0", "[::]:0", "192.0.2.1:3737", "invalid"] { - let output = run(&[], &["--listen", address]); - assert!(!output.status.success()); - assert!(!output.stderr.is_empty()); - } -} - -#[test] -fn tcp_listener_accepts_framed_sessions_and_survives_invalid_framing() { - let mut child = Command::new(env!("CARGO_BIN_EXE_runa-dap")) - .args(["--listen", "127.0.0.1:0"]) - .env_remove("RUNA_CONNECT") - .env_remove("RUNA_DAP_LISTEN") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let mut line = String::new(); - let mut diagnostics = BufReader::new(child.stderr.take().unwrap()); - diagnostics.read_line(&mut line).unwrap(); - let address = line.trim().strip_prefix("runa-dap: listening on ").unwrap(); - for malformed in [true, false, false] { - let mut socket = TcpStream::connect(address).unwrap(); - socket - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - if malformed { - socket - .write_all(b"Content-Length: 1048577\r\n\r\n") - .unwrap(); - let mut reader = BufReader::new(socket); - assert_eq!(read_frame(&mut reader).unwrap().unwrap()["event"], "output"); - assert_eq!( - read_frame(&mut reader).unwrap().unwrap()["event"], - "terminated" - ); - } else { - write_frame( - &mut socket, - &json!({ - "seq": 1, "type": "request", "command": "initialize" - }), - ) - .unwrap(); - let mut reader = BufReader::new(socket.try_clone().unwrap()); - assert_eq!(read_frame(&mut reader).unwrap().unwrap()["success"], true); - write_frame( - &mut socket, - &json!({ - "seq": 2, "type": "request", "command": "disconnect" - }), - ) - .unwrap(); - assert_eq!(read_frame(&mut reader).unwrap().unwrap()["success"], true); - assert_eq!( - read_frame(&mut reader).unwrap().unwrap()["event"], - "terminated" - ); - } - } - })); - child.kill().unwrap(); - child.wait().unwrap(); - result.unwrap(); -} diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index e837dd1..93b0edd 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -8,6 +8,7 @@ clap.workspace = true futures-util = { version = "0.3", features = ["sink"] } runa-actor = { path = "../runa-actor" } runa-code = { path = "../runa-code" } +runa-dap = { path = "../runa-dap" } runa-image-api = { path = "../runa-image-api" } runa-protocol = { path = "../runa-protocol" } serde_json = "1" diff --git a/crates/runa/src/dap.rs b/crates/runa/src/dap.rs new file mode 100644 index 0000000..ecaf328 --- /dev/null +++ b/crates/runa/src/dap.rs @@ -0,0 +1,175 @@ +//! Server-hosted loopback TCP DAP sessions bridged to the single image worker. + +use std::io::{self, BufRead, BufReader, BufWriter, Read}; +use std::net::{Shutdown, SocketAddr, TcpStream}; +use std::sync::mpsc::{self, RecvTimeoutError}; +use std::time::{Duration, Instant}; + +use runa_dap::{Adapter, Connection, framing}; +use runa_protocol::{Command, PROTOCOL_VERSION, Reply, Request}; +use serde_json::Value as Json; +use tokio::sync::OwnedSemaphorePermit; + +use crate::network::{Dispatch, Responder}; + +pub(crate) const MAX_SESSIONS: usize = 8; +const POLL_INTERVAL: Duration = Duration::from_millis(150); +const FRAME_TIMEOUT: Duration = Duration::from_secs(30); +const WRITE_TIMEOUT: Duration = Duration::from_secs(30); +const DISPATCH_TIMEOUT: Duration = Duration::from_secs(30); + +/// Run one DAP session on its own thread; the permit bounds concurrent sessions. +pub(crate) fn spawn_session( + stream: tokio::net::TcpStream, + address: SocketAddr, + requests: tokio::sync::mpsc::Sender, + permit: OwnedSemaphorePermit, +) -> io::Result<()> { + let stream = stream.into_std()?; + stream.set_nonblocking(false)?; + std::thread::Builder::new() + .name("runa-dap-session".to_owned()) + .spawn(move || { + let _permit = permit; + if let Err(error) = session(stream, requests) { + eprintln!("runa: dap client {address}: {error}"); + } + })?; + Ok(()) +} + +/// Dispatches DAP commands to the image worker and waits for its reply. +/// +/// This blocks the calling session thread and must never run on the worker runtime. +struct WorkerConnection { + requests: tokio::sync::mpsc::Sender, + next_request: u64, +} + +impl Connection for WorkerConnection { + fn request(&mut self, command: Command) -> Result { + self.next_request += 1; + let (respond, response) = mpsc::sync_channel(1); + let request = Request { + version: PROTOCOL_VERSION, + request_id: format!("dap-{}", self.next_request), + command, + }; + self.requests + .blocking_send(Dispatch { + request, + respond: Responder::Blocking(respond), + }) + .map_err(|_| "Image worker stopped".to_owned())?; + match response.recv_timeout(DISPATCH_TIMEOUT) { + Ok(response) => response.result.map_err(|error| error.message), + Err(RecvTimeoutError::Timeout) => { + Err("Image worker did not respond in time".to_owned()) + } + Err(RecvTimeoutError::Disconnected) => Err("Image worker stopped".to_owned()), + } + } +} + +/// Allows idle sessions while bounding the time taken by each started frame. +struct DeadlineReader { + stream: TcpStream, + deadline: Option, +} + +impl Read for DeadlineReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + let timeout = match self.deadline { + None => None, + Some(deadline) => { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "incomplete DAP frame timed out", + )); + } + Some(remaining) + } + }; + self.stream.set_read_timeout(timeout)?; + self.stream.read(buffer).map_err(|error| { + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) { + io::Error::new(io::ErrorKind::TimedOut, "incomplete DAP frame timed out") + } else { + error + } + }) + } +} + +type Frame = io::Result>; + +fn read_frames(stream: TcpStream, sender: mpsc::SyncSender) { + let mut reader = BufReader::new(DeadlineReader { + stream, + deadline: None, + }); + loop { + reader.get_mut().deadline = None; + let result = match reader.fill_buf() { + Ok([]) => Ok(None), + Ok(_) => { + reader.get_mut().deadline = Some(Instant::now() + FRAME_TIMEOUT); + framing::read_frame(&mut reader) + } + Err(error) => Err(error), + }; + let finished = !matches!(&result, Ok(Some(_))); + if sender.send(result).is_err() || finished { + break; + } + } +} + +fn session(stream: TcpStream, requests: tokio::sync::mpsc::Sender) -> io::Result<()> { + stream.set_write_timeout(Some(WRITE_TIMEOUT))?; + let (sender, receiver) = mpsc::sync_channel(16); + let reader = { + let stream = stream.try_clone()?; + std::thread::Builder::new() + .name("runa-dap-reader".to_owned()) + .spawn(move || read_frames(stream, sender))? + }; + let mut output = BufWriter::new(stream.try_clone()?); + let mut adapter = Adapter::new(WorkerConnection { + requests, + next_request: 0, + }); + let result = (|| { + while !adapter.is_closed() { + let messages = match receiver.recv_timeout(POLL_INTERVAL) { + Ok(Ok(Some(message))) => adapter.handle(message), + Ok(Ok(None)) | Err(RecvTimeoutError::Disconnected) => break, + Ok(Err(error)) => { + for message in adapter.terminate(&format!("Invalid DAP framing: {error}")) { + framing::write_frame(&mut output, &message)?; + } + return Err(error); + } + Err(RecvTimeoutError::Timeout) => adapter.poll(), + }; + for message in messages { + framing::write_frame(&mut output, &message)?; + } + // Poll even under a steady stream of input; requests cannot starve lifecycle checks. + for message in adapter.poll() { + framing::write_frame(&mut output, &message)?; + } + } + Ok(()) + })(); + // Closing the socket unblocks the reader; dropping the receiver unblocks its sends. + drop(receiver); + let _ = stream.shutdown(Shutdown::Both); + let _ = reader.join(); + result +} diff --git a/crates/runa/src/lib.rs b/crates/runa/src/lib.rs index 08f8ec7..b5046bb 100644 --- a/crates/runa/src/lib.rs +++ b/crates/runa/src/lib.rs @@ -4,9 +4,10 @@ use std::io; use std::net::{SocketAddr, TcpListener}; use std::num::NonZeroUsize; +mod dap; mod network; mod storage; -pub use network::serve_websocket; +pub use network::{serve, serve_websocket}; use runa_actor::{ActorRef, Event, Status, SystemMessage}; use runa_image_api::ImageApi; diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index 5363cd8..6eb7678 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -3,7 +3,7 @@ use std::net::SocketAddr; use std::process::ExitCode; use clap::Parser; -use runa::{Server, bind_loopback_listener, serve_websocket}; +use runa::{Server, bind_loopback_listener, serve}; use runa_protocol::{Command, PROTOCOL_VERSION, Request}; #[derive(Debug, Parser)] @@ -12,6 +12,9 @@ struct Args { /// Loopback WebSocket listen address (port 0 selects an ephemeral port). #[arg(long, env = "RUNA_LISTEN", default_value = "127.0.0.1:3737")] listen: SocketAddr, + /// Loopback Debug Adapter Protocol TCP listen address (port 0 selects an ephemeral port). + #[arg(long, env = "RUNA_DAP_LISTEN", default_value = "127.0.0.1:4711")] + dap_listen: SocketAddr, /// Restore a server-local image snapshot instead of loading bundled sources. #[arg(long, env = "RUNA_RESTORE")] restore: Option, @@ -42,9 +45,13 @@ fn initial_image(restore: Option<&str>) -> io::Result { fn main() -> ExitCode { let args = Args::parse(); - match bind_loopback_listener(args.listen) - .and_then(|listener| serve_websocket(listener, initial_image(args.restore.as_deref())?)) - { + // Bind both endpoints and load the image before advertising either listener. + let started = (|| { + let websocket = bind_loopback_listener(args.listen)?; + let dap = bind_loopback_listener(args.dap_listen)?; + serve(websocket, dap, initial_image(args.restore.as_deref())?) + })(); + match started { Ok(()) => ExitCode::SUCCESS, Err(error) => { eprintln!("runa: {error}"); @@ -65,6 +72,23 @@ mod tests { .find(|argument| argument.get_id() == "listen") .unwrap(); assert_eq!(listen.get_default_values(), ["127.0.0.1:3737"]); + let dap_listen = command + .get_arguments() + .find(|argument| argument.get_id() == "dap_listen") + .unwrap(); + assert_eq!(dap_listen.get_default_values(), ["127.0.0.1:4711"]); + assert_eq!( + dap_listen.get_env(), + Some(std::ffi::OsStr::new("RUNA_DAP_LISTEN")) + ); + assert_eq!( + Args::try_parse_from(["runa", "--dap-listen", "[::1]:0"]) + .unwrap() + .dap_listen, + "[::1]:0".parse::().unwrap() + ); + assert!(Args::try_parse_from(["runa", "--dap-listen"]).is_err()); + assert!(Args::try_parse_from(["runa", "--dap-listen", "invalid"]).is_err()); assert_eq!( Args::try_parse_from(["runa", "--listen", "[::1]:4000"]) .unwrap() diff --git a/crates/runa/src/network.rs b/crates/runa/src/network.rs index f0b07ef..20c093f 100644 --- a/crates/runa/src/network.rs +++ b/crates/runa/src/network.rs @@ -1,12 +1,13 @@ use std::io; use std::net::TcpListener; +use std::sync::Arc; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; use runa_protocol::{ Command, ImageEvent, Notification, PROTOCOL_VERSION, Request, Response, WEBSOCKET_SUBPROTOCOL, }; -use tokio::sync::{broadcast, mpsc, oneshot, watch}; +use tokio::sync::{Semaphore, broadcast, mpsc, oneshot, watch}; use tokio::task::JoinSet; use tokio::time::timeout; use tokio_tungstenite::tungstenite::handshake::server::{ @@ -15,38 +16,78 @@ use tokio_tungstenite::tungstenite::handshake::server::{ use tokio_tungstenite::tungstenite::http::{HeaderValue, StatusCode}; use tokio_tungstenite::tungstenite::{Message, protocol::WebSocketConfig}; -use crate::{MAX_MESSAGE_BYTES, Server}; +use crate::{MAX_MESSAGE_BYTES, Server, dap}; const MAX_NETWORK_CLIENTS: usize = 16; const WRITE_TIMEOUT: Duration = Duration::from_secs(30); -/// Keep all image dispatch on one thread, with bounded per-client I/O tasks. +/// Serve only the WebSocket protocol, with all image dispatch on one thread. pub fn serve_websocket(listener: TcpListener, server: Server) -> io::Result<()> { - if !listener.local_addr()?.ip().is_loopback() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "network listener must be bound to a loopback address", - )); + run_server(listener, None, server) +} + +/// Serve the WebSocket protocol and loopback TCP DAP sessions from one image. +/// +/// Both listeners must already be bound to loopback addresses. DAP requests +/// are dispatched through the same single-worker request channel as WebSocket +/// requests, so successful DAP mutations also publish image invalidations. +pub fn serve(websocket: TcpListener, dap: TcpListener, server: Server) -> io::Result<()> { + run_server(websocket, Some(dap), server) +} + +fn run_server(websocket: TcpListener, dap: Option, server: Server) -> io::Result<()> { + for listener in std::iter::once(&websocket).chain(dap.as_ref()) { + if !listener.local_addr()?.ip().is_loopback() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "network listener must be bound to a loopback address", + )); + } + listener.set_nonblocking(true)?; } - listener.set_nonblocking(true)?; tokio::runtime::Builder::new_current_thread() .enable_all() .build()? - .block_on(serve(listener, server)) + .block_on(run(websocket, dap, server)) } -struct Dispatch { - request: Request, - respond: oneshot::Sender, +pub(crate) enum Responder { + Async(oneshot::Sender), + Blocking(std::sync::mpsc::SyncSender), } -async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { +pub(crate) struct Dispatch { + pub(crate) request: Request, + pub(crate) respond: Responder, +} + +async fn accept( + listener: Option<&tokio::net::TcpListener>, +) -> io::Result<(tokio::net::TcpStream, std::net::SocketAddr)> { + match listener { + Some(listener) => listener.accept().await, + None => std::future::pending().await, + } +} + +async fn run( + listener: TcpListener, + dap_listener: Option, + mut server: Server, +) -> io::Result<()> { let listener = tokio::net::TcpListener::from_std(listener)?; + let dap_listener = dap_listener + .map(tokio::net::TcpListener::from_std) + .transpose()?; eprintln!("runa: listening on {}", listener.local_addr()?); + if let Some(dap_listener) = &dap_listener { + eprintln!("runa: dap listening on {}", dap_listener.local_addr()?); + } let (requests, mut pending) = mpsc::channel::(MAX_NETWORK_CLIENTS); let (changes, _) = watch::channel(0_u64); let mut revision = 0_u64; let mut clients = JoinSet::new(); + let dap_sessions = Arc::new(Semaphore::new(dap::MAX_SESSIONS)); let (logs, _) = broadcast::channel::(1024); let log_sender = logs.clone(); server.image.set_log_sink(move |entry| { @@ -70,6 +111,18 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { } }); } + accepted = accept(dap_listener.as_ref()) => { + let (stream, address) = accepted?; + let Ok(permit) = dap_sessions.clone().try_acquire_owned() else { + eprintln!("runa: rejecting dap client {address}; session limit reached"); + continue; + }; + // Sessions block on their sockets, so they run on dedicated threads + // and reach the image only through the bounded request channel. + if let Err(error) = dap::spawn_session(stream, address, requests.clone(), permit) { + eprintln!("runa: dap client {address}: {error}"); + } + } Some(dispatch) = pending.recv() => { let mutates = !matches!(dispatch.request.command, Command::GetBehavior { .. } | Command::GetBehaviors | @@ -80,7 +133,11 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { let response = server.dispatch(dispatch.request); let changed = mutates && response.result.is_ok(); // Dispatch may commit even if the initiating client has disconnected. - let _ = dispatch.respond.send(response); + match dispatch.respond { + Responder::Async(respond) => { let _ = respond.send(response); } + // The sync channel has capacity for the single reply, so this never blocks. + Responder::Blocking(respond) => { let _ = respond.try_send(response); } + } if changed { revision = revision.checked_add(1) .ok_or_else(|| io::Error::other("image notification revisions exhausted"))?; @@ -158,7 +215,7 @@ async fn client( }; let receiver = requested_logs.as_ref().filter(|(_, follow)| *follow).map(|_| logs.subscribe()); let (respond, response) = oneshot::channel(); - requests.send(Dispatch { request, respond }).await?; + requests.send(Dispatch { request, respond: Responder::Async(respond) }).await?; let response = response.await?; if let (Some((actors, _)), Ok(runa_protocol::Reply::Logs { next_sequence, .. })) = (requested_logs, &response.result) { subscription = receiver; diff --git a/crates/runa-dap/tests/adapter.rs b/crates/runa/tests/dap_adapter.rs similarity index 100% rename from crates/runa-dap/tests/adapter.rs rename to crates/runa/tests/dap_adapter.rs diff --git a/crates/runa/tests/dap_server.rs b/crates/runa/tests/dap_server.rs new file mode 100644 index 0000000..13ac3dc --- /dev/null +++ b/crates/runa/tests/dap_server.rs @@ -0,0 +1,531 @@ +//! Process-level tests for the server-hosted loopback TCP DAP endpoint. + +use std::io::{BufRead, BufReader, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::process::{Child, Command as ProcessCommand, Stdio}; +use std::time::{Duration, Instant}; + +use runa_client::WebSocketClient; +use runa_dap::framing::{read_frame, write_frame}; +use runa_protocol::{ActorStatusDto, Command, Reply, Value}; +use serde_json::{Value as Json, json}; +use tokio_tungstenite::tungstenite::{self, client::IntoClientRequest}; + +const SOURCE: &str = r#"fn helper(value) { + value + 1 +} +pub fn handle(ctx, state, sender, message) { + let next = helper(state); + next + message +} +"#; + +struct ServerProcess(Child); + +impl Drop for ServerProcess { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn runa() -> ProcessCommand { + let mut command = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")); + command + .env_remove("RUNA_LISTEN") + .env_remove("RUNA_DAP_LISTEN") + .env_remove("RUNA_RESTORE") + .stdin(Stdio::null()) + .stdout(Stdio::null()); + command +} + +fn start(mut command: ProcessCommand) -> (ServerProcess, SocketAddr, SocketAddr) { + let mut server = ServerProcess(command.stderr(Stdio::piped()).spawn().unwrap()); + let mut stderr = BufReader::new(server.0.stderr.take().unwrap()); + let mut line = |prefix: &str| -> SocketAddr { + let mut text = String::new(); + stderr.read_line(&mut text).unwrap(); + text.strip_prefix(prefix) + .unwrap_or_else(|| panic!("expected {prefix:?}, got {text:?}")) + .trim() + .parse() + .unwrap() + }; + let websocket = line("runa: listening on "); + let dap = line("runa: dap listening on "); + std::thread::spawn(move || std::io::copy(&mut stderr, &mut std::io::sink())); + (server, websocket, dap) +} + +fn start_default() -> (ServerProcess, SocketAddr, SocketAddr) { + let mut command = runa(); + command.args(["--listen", "127.0.0.1:0", "--dap-listen", "127.0.0.1:0"]); + start(command) +} + +fn spawn_actor(client: &mut WebSocketClient, path: &str) -> String { + request( + client, + Command::WriteSource { + path: path.to_owned(), + expected_source_id: None, + source: SOURCE.to_owned(), + }, + ); + let Reply::ActorSpawned { actor_ref } = request( + client, + Command::Spawn { + behavior_name: path.to_owned(), + mailbox_capacity: 32, + state: Value::Integer(10), + }, + ) else { + panic!("expected spawned actor") + }; + actor_ref +} + +fn request(client: &mut WebSocketClient, command: Command) -> Reply { + client.request(command).unwrap().result.unwrap() +} + +fn inspect(client: &mut WebSocketClient, actor: &str) -> runa_protocol::ActorSnapshotDto { + let Reply::Actor(actor) = request( + client, + Command::InspectActor { + actor_ref: actor.to_owned(), + }, + ) else { + panic!("expected actor") + }; + actor +} + +struct Dap { + reader: BufReader, + writer: TcpStream, + seq: u64, + pending: Vec, +} + +impl Dap { + fn connect(address: SocketAddr) -> Self { + let writer = TcpStream::connect(address).unwrap(); + writer + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + Self { + reader: BufReader::new(writer.try_clone().unwrap()), + writer, + seq: 0, + pending: Vec::new(), + } + } + + fn read(&mut self) -> Json { + read_frame(&mut self.reader) + .unwrap() + .expect("DAP session closed") + } + + fn send(&mut self, command: &str, arguments: Json) -> Json { + self.seq += 1; + let seq = self.seq; + write_frame( + &mut self.writer, + &json!({"seq": seq, "type": "request", "command": command, "arguments": arguments}), + ) + .unwrap(); + loop { + let message = self.read(); + if message["type"] == "response" && message["request_seq"] == seq { + assert_eq!(message["command"], command); + return message; + } + self.pending.push(message); + } + } + + fn ok(&mut self, command: &str, arguments: Json) -> Json { + let response = self.send(command, arguments); + assert_eq!(response["success"], true, "{response}"); + response["body"].clone() + } + + fn event(&mut self, matches: impl Fn(&Json) -> bool) -> Json { + if let Some(index) = self.pending.iter().position(&matches) { + return self.pending.remove(index); + } + loop { + let message = self.read(); + if matches(&message) { + return message; + } + self.pending.push(message); + } + } + + fn attach_image(&mut self) { + let capabilities = self.ok("initialize", json!({"adapterID": "runa"})); + assert_eq!(capabilities["supportsConfigurationDoneRequest"], true); + self.attach_image_after_initialize(); + } + + fn attach_image_after_initialize(&mut self) { + self.ok( + "attach", + json!({ + "image": true, + "actorSourceBaseUri": "runa-actor://session", + "workingSourceBaseUri": "runa-image://localhost/source" + }), + ); + self.event(|message| message["event"] == "initialized"); + self.ok("configurationDone", json!({})); + } + + fn thread(&mut self, actor: &str) -> u64 { + self.ok("threads", json!({}))["threads"] + .as_array() + .unwrap() + .iter() + .find(|thread| thread["name"] == format!("@{actor}")) + .unwrap_or_else(|| panic!("missing thread for {actor}"))["id"] + .as_u64() + .unwrap() + } + + fn assert_closed(&mut self) { + loop { + match read_frame(&mut self.reader) { + Ok(Some(_)) => {} + Ok(None) | Err(_) => return, + } + } + } +} + +fn watcher(address: SocketAddr) -> tungstenite::WebSocket { + let mut request = format!("ws://{address}/").into_client_request().unwrap(); + request + .headers_mut() + .insert("sec-websocket-protocol", "runa-debug".parse().unwrap()); + let stream = TcpStream::connect(address).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + tungstenite::client(request, stream).unwrap().0 +} + +fn revision(socket: &mut tungstenite::WebSocket) -> u64 { + let message = socket.read().unwrap(); + let event: runa_protocol::Notification = + serde_json::from_str(message.to_text().unwrap()).unwrap(); + let runa_protocol::ImageEvent::ImageChanged { revision } = event.event else { + panic!("expected image invalidation") + }; + revision +} + +#[test] +fn dap_sessions_share_the_websocket_image_and_publish_invalidations() { + let (_server, websocket, dap) = start_default(); + let mut client = WebSocketClient::connect(websocket).unwrap(); + let actor = spawn_actor(&mut client, "tests/handler"); + let mut changes = watcher(websocket); + + let mut session = Dap::connect(dap); + session.attach_image(); + let thread = session.thread(&actor); + let source = json!({"path": format!("runa-actor://session/{actor}/behavior.rn")}); + let breakpoints = session.ok( + "setBreakpoints", + json!({"source": source, "breakpoints": [{"line": 5}]}), + ); + assert_eq!(breakpoints["breakpoints"][0]["verified"], true); + // The DAP mutation went through the shared worker, so WebSocket clients are told. + let first_revision = revision(&mut changes); + let Reply::ExecutionInspected(execution) = request( + &mut client, + Command::InspectExecution { + actor_ref: actor.clone(), + }, + ) else { + panic!("expected execution") + }; + assert_eq!(execution.debugger.breakpoints.len(), 1); + assert_eq!( + execution.debugger.breakpoints[0].line, 4, + "protocol lines are zero-based" + ); + + // A second session has independent DAP state and does not disturb the first. + let mut other = Dap::connect(dap); + other.ok("initialize", json!({})); + assert_eq!(session.send("initialize", json!({}))["success"], false); + other.attach_image_after_initialize(); + other.thread(&actor); + other.ok("disconnect", json!({})); + other.assert_closed(); + + request( + &mut client, + Command::Send { + actor_ref: actor.clone(), + message: Value::Integer(2), + }, + ); + request( + &mut client, + Command::Tick { + ticks: 1, + instruction_budget: 100_000, + }, + ); + let stopped = session.event(|message| message["event"] == "stopped"); + assert_eq!(stopped["body"]["reason"], "breakpoint", "{stopped}"); + assert_eq!(stopped["body"]["threadId"], thread); + assert_eq!(inspect(&mut client, &actor).status, ActorStatusDto::Paused); + assert!(revision(&mut changes) > first_revision); + + let trace = session.ok("stackTrace", json!({"threadId": thread})); + assert!(trace["totalFrames"].as_u64().unwrap() > 0, "{trace}"); + assert_eq!(trace["stackFrames"][0]["line"], 5, "{trace}"); + session.ok("stepIn", json!({"threadId": thread})); + let stepped = session.event(|message| message["event"] == "stopped"); + assert_eq!(stepped["body"]["reason"], "step", "{stepped}"); + let trace = session.ok("stackTrace", json!({"threadId": thread})); + assert!( + trace["stackFrames"].as_array().unwrap().len() > 1, + "{trace}" + ); + session.ok("stepOut", json!({"threadId": thread})); + session.event(|message| message["event"] == "stopped"); + session.ok("next", json!({"threadId": thread})); + session.event(|message| message["event"] == "stopped"); + + session.ok( + "setBreakpoints", + json!({"source": {"path": format!("runa-actor://session/{actor}/behavior.rn")}, "breakpoints": []}), + ); + session.ok("continue", json!({"threadId": thread})); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + request( + &mut client, + Command::Tick { + ticks: 1, + instruction_budget: 100_000, + }, + ); + let snapshot = inspect(&mut client, &actor); + if snapshot.state == Value::Integer(13) { + assert_ne!(snapshot.status, ActorStatusDto::Paused); + break; + } + assert!(Instant::now() < deadline, "{snapshot:?}"); + std::thread::sleep(Duration::from_millis(20)); + } + session.ok("disconnect", json!({})); + session.assert_closed(); + assert!(matches!( + request(&mut client, Command::ListActors), + Reply::Actors(_) + )); +} + +#[test] +fn actor_mode_sessions_attach_through_the_server() { + let (_server, websocket, dap) = start_default(); + let mut client = WebSocketClient::connect(websocket).unwrap(); + let actor = spawn_actor(&mut client, "tests/handler"); + let mut session = Dap::connect(dap); + session.ok("initialize", json!({})); + session.ok( + "attach", + json!({ + "actorRef": actor, + "behaviorSourceUri": format!("runa-actor://session/{actor}/behavior.rn"), + "workingSourceUri": "runa-image://localhost/source/tests/handler.rn" + }), + ); + session.event(|message| message["event"] == "initialized"); + session.ok("configurationDone", json!({})); + let stopped = session.event(|message| message["event"] == "stopped"); + assert_eq!(stopped["body"]["reason"], "pause"); + assert_eq!(inspect(&mut client, &actor).status, ActorStatusDto::Paused); + assert_eq!( + session.ok("source", json!({"sourceReference": 1}))["content"], + SOURCE + ); + session.ok("continue", json!({"threadId": 1})); + assert_ne!(inspect(&mut client, &actor).status, ActorStatusDto::Paused); + session.ok("disconnect", json!({})); + session.event(|message| message["event"] == "terminated"); + session.assert_closed(); +} + +#[test] +fn malformed_dap_clients_are_isolated() { + let (_server, websocket, dap) = start_default(); + let mut healthy = Dap::connect(dap); + healthy.ok("initialize", json!({})); + + // A stalled partial frame does not block other sessions or the worker. + let mut stalled = TcpStream::connect(dap).unwrap(); + stalled.write_all(b"Content-Len").unwrap(); + + for bytes in [ + &b"Content-Length: 1048577\r\n\r\n"[..], + b"Content-Length: 2\r\n\r\nno", + b"Content-Length: 2\n\n{}", + ] { + let mut malformed = Dap::connect(dap); + malformed.writer.write_all(bytes).unwrap(); + let output = malformed.read(); + assert_eq!(output["event"], "output"); + assert!( + output["body"]["output"] + .as_str() + .unwrap() + .contains("Invalid DAP framing") + ); + assert_eq!(malformed.read()["event"], "terminated"); + malformed.assert_closed(); + } + let mut invalid = Dap::connect(dap); + write_frame(&mut invalid.writer, &json!({"seq": 1, "type": "event"})).unwrap(); + assert_eq!(invalid.read()["event"], "output"); + assert_eq!(invalid.read()["event"], "terminated"); + invalid.assert_closed(); + + healthy.ok("attach", json!({"image": true, "actorSourceBaseUri": "runa-actor://session", "workingSourceBaseUri": "runa-image://localhost/source"})); + healthy.event(|message| message["event"] == "initialized"); + healthy.ok("configurationDone", json!({})); + assert_eq!(healthy.ok("threads", json!({}))["threads"], json!([])); + let mut client = WebSocketClient::connect(websocket).unwrap(); + assert!(matches!( + request(&mut client, Command::ListActors), + Reply::Actors(_) + )); + drop(stalled); +} + +#[test] +fn dap_session_limit_is_bounded_and_slots_are_released() { + let (_server, _websocket, dap) = start_default(); + let mut sessions: Vec = (0..8) + .map(|_| { + let mut session = Dap::connect(dap); + session.ok("initialize", json!({})); + session + }) + .collect(); + let mut rejected = Dap::connect(dap); + let _ = write_frame( + &mut rejected.writer, + &json!({"seq": 1, "type": "request", "command": "initialize"}), + ); + assert!(!matches!(read_frame(&mut rejected.reader), Ok(Some(_)))); + let mut released = sessions.pop().unwrap(); + released.ok("disconnect", json!({})); + released.assert_closed(); + drop(released); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let mut session = Dap::connect(dap); + let _ = write_frame( + &mut session.writer, + &json!({"seq": 1, "type": "request", "command": "initialize"}), + ); + if let Ok(Some(response)) = read_frame(&mut session.reader) { + assert_eq!(response["success"], true); + break; + } + assert!(Instant::now() < deadline, "DAP slot was not released"); + std::thread::sleep(Duration::from_millis(50)); + } + for session in &mut sessions { + let response = session.send("initialize", json!({})); + assert!( + response["message"] + .as_str() + .unwrap() + .contains("already initialized") + ); + } +} + +#[test] +fn dap_listener_cli_env_precedence_and_help() { + let mut command = runa(); + command + .args(["--listen", "127.0.0.1:0", "--dap-listen", "127.0.0.1:0"]) + .env("RUNA_DAP_LISTEN", "invalid"); + let (_cli, _, dap) = start(command); + Dap::connect(dap).ok("initialize", json!({})); + + let mut command = runa(); + command + .env("RUNA_LISTEN", "127.0.0.1:0") + .env("RUNA_DAP_LISTEN", "127.0.0.1:0"); + let (_env, _, dap) = start(command); + assert_ne!(dap.port(), 4711); + Dap::connect(dap).ok("initialize", json!({})); + + let help = runa() + .arg("--help") + .stdout(Stdio::piped()) + .output() + .unwrap(); + assert!(help.status.success()); + let help = String::from_utf8(help.stdout).unwrap(); + for text in [ + "--dap-listen", + "RUNA_DAP_LISTEN", + "127.0.0.1:4711", + "--listen", + ] { + assert!(help.contains(text), "{help}"); + } +} + +#[test] +fn invalid_or_unavailable_listeners_fail_before_advertising() { + let occupied = TcpListener::bind("127.0.0.1:0").unwrap(); + let occupied = occupied.local_addr().unwrap().to_string(); + let ephemeral = "127.0.0.1:0"; + let cases: Vec<(Vec<&str>, Option<&str>)> = vec![ + (vec!["--listen", ephemeral, "--dap-listen", "invalid"], None), + (vec!["--listen", ephemeral, "--dap-listen"], None), + ( + vec!["--listen", ephemeral, "--dap-listen", "0.0.0.0:0"], + None, + ), + (vec!["--listen", ephemeral, "--dap-listen", "[::]:0"], None), + ( + vec!["--listen", ephemeral, "--dap-listen", "192.0.2.1:4711"], + None, + ), + (vec!["--listen", ephemeral, "--dap-listen", &occupied], None), + (vec!["--listen", &occupied, "--dap-listen", ephemeral], None), + (vec!["--listen", ephemeral], Some("0.0.0.0:0")), + (vec!["--listen", ephemeral], Some("invalid")), + ]; + for (args, env) in cases { + let mut command = runa(); + command.args(&args); + if let Some(env) = env { + command.env("RUNA_DAP_LISTEN", env); + } + let output = command.stderr(Stdio::piped()).output().unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(!output.status.success(), "{args:?} {env:?}: {stderr}"); + assert!(!stderr.is_empty(), "{args:?} {env:?}"); + assert!( + !stderr.contains("listening on"), + "{args:?} {env:?}: {stderr}" + ); + } +} diff --git a/crates/runa/tests/local_client.rs b/crates/runa/tests/local_client.rs index b5e5c1f..850f22c 100644 --- a/crates/runa/tests/local_client.rs +++ b/crates/runa/tests/local_client.rs @@ -49,6 +49,7 @@ fn start_server_with_restore( ) -> (ServerProcess, SocketAddr) { let mut command = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")); command + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .args(args) .env("RUNA_LISTEN", listen_env) .env_remove("RUNA_RESTORE"); @@ -90,6 +91,7 @@ fn server_clap_supports_env_overrides_help_and_explicit_errors() { vec!["--listen", "0.0.0.0:0"], ] { let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .args(args) .env_remove("RUNA_LISTEN") .env_remove("RUNA_RESTORE") @@ -99,6 +101,7 @@ fn server_clap_supports_env_overrides_help_and_explicit_errors() { assert!(!output.stderr.is_empty()); } let help = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .arg("--help") .env_remove("RUNA_LISTEN") .env_remove("RUNA_RESTORE") @@ -244,6 +247,7 @@ fn startup_restore_failures_exit_without_advertising_a_live_image() { "", ] { let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .args(["--listen", "127.0.0.1:0", "--restore", file]) .env_remove("RUNA_LISTEN") .env_remove("RUNA_RESTORE") @@ -255,6 +259,7 @@ fn startup_restore_failures_exit_without_advertising_a_live_image() { assert!(!error.contains("listening on"), "{error}"); } let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .args(["--listen", "127.0.0.1:0"]) .env("RUNA_RESTORE", missing) .output() @@ -534,6 +539,7 @@ impl Drop for ServerProcess { #[test] fn websocket_server_accepts_local_protocol_requests() { let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .env("RUNA_DAP_LISTEN", "127.0.0.1:0") .args(["--listen", "127.0.0.1:0"]) .env_remove("RUNA_RESTORE") .stdout(Stdio::null()) diff --git a/docs/architecture.md b/docs/architecture.md index a4b3412..feffbf1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -240,7 +240,7 @@ unavailable. ## 8. Unified Programmatic Image API -The running image exposes a versioned semantic API, used by graphical tools, CLI tools, tests, automation, coding agents, and the VS Code extension. External clients use WebSockets to reach the same running image; there is no stdio transport. Client location does not change actor placement or message-passing semantics. +The running image exposes a versioned semantic API, used by graphical tools, CLI tools, tests, automation, coding agents, and the VS Code extension. External image API clients use WebSockets; the server also hosts a loopback TCP DAP endpoint for debug clients. Both dispatch through the same image worker, with no stdio transport. Client location does not change actor placement or message-passing semantics. Suggested operation families (provisional names): @@ -578,9 +578,10 @@ exact-loaded-source documents, including old code during pending replacements. Await locations represent the last safe snapshot; native/hidden async frames, locals and post-mortem stacks remain unimplemented. Actor-specific implementation-bound line breakpoints and bounded step-into/over/out now preserve active -drafts/effects and use exact loaded-source mappings. A standalone Rust `runa-dap` -server speaks DAP over stdio and connects to the image over loopback WebSocket. -The VS Code extension launches it as an executable adapter. Every actor exposes +drafts/effects and use exact loaded-source mappings. The image server hosts the +Rust `runa-dap` library as a loopback TCP DAP endpoint, dispatching through the +same single-worker request queue as the WebSocket API. The VS Code extension +connects directly; there is no standalone adapter executable. Every actor exposes `behavior.rn`; its gutter breakpoints bind only to that actor's exact loaded code. The attach-only adapter exposes one actor as one VS Code debug thread with exact loaded sources, gutter breakpoints, call stacks, pause/resume, and bounded toolbar diff --git a/docs/protocol.md b/docs/protocol.md index e778332..8195731 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -208,7 +208,7 @@ configuration and no reason. `stop_reason` is nullable and uses the same reason strings as step replies. Installation clears breakpoints; restart/cancel retain them for unchanged loaded code and clear stop reasons. Debugger state is actor local, shared by clients, and in-memory; it is not owned by a client connection. -There are no conditions or hit counts. The standalone `runa-dap` adapter maps +There are no conditions or hit counts. The server-hosted `runa-dap` library maps DAP gutter breakpoints to these actor-local controls. System message shape is `{kind: "stop"|"kill"|"restart"|"pause"|"resume"|"reload"}` @@ -224,21 +224,17 @@ restarted. Reload/become defer installation to the next handler boundary. Normal scheduling advances through `tick`; the explicit debugger step operation advances only its selected paused actor. -### Standalone Debug Adapter Protocol server +### Server-hosted Debug Adapter Protocol endpoint -`runa-dap --connect 127.0.0.1:3737` speaks standard Content-Length-framed DAP -on stdin/stdout and connects to an existing trusted loopback image using the -WebSocket operations above. Diagnostics go to stderr. It does not start an -image or tick the scheduler implicitly. Stdio mode opens no DAP network listener. - -To accept DAP over TCP instead of stdio, use -`runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737`. -`RUNA_DAP_LISTEN` also selects the DAP bind address; `--listen` overrides it. -Only loopback addresses are accepted because the API is unauthenticated. -Port `0` chooses an available port, reported on stderr. TCP uses the same -Content-Length framing, handles one session at a time, and accepts a new -session after disconnect or a malformed session. Without `--listen` or -`RUNA_DAP_LISTEN`, the existing stdio behavior is unchanged. +The image server speaks standard Content-Length-framed DAP over loopback TCP, +defaulting to `127.0.0.1:4711`. Set `runa --dap-listen 127.0.0.1:4712` or +`RUNA_DAP_LISTEN` to change the bind address; the CLI flag takes precedence. +Port `0` chooses an available port, reported on stderr as +`runa: dap listening on ...`. Only loopback addresses are accepted because the +API is unauthenticated. DAP requests dispatch through the same single image +worker as WebSocket requests; no second image or adapter process is created. +Debugger controls do not tick the scheduler implicitly. The standalone binary +and stdio DAP transport have been removed. An attach request supplies `actorRef`, optionally `behaviorSourceUri` (the actor's `behavior.rn` editor URI) and `workingSourceUri` (its shared image source diff --git a/mise.toml b/mise.toml index b8a9b78..bbed16d 100644 --- a/mise.toml +++ b/mise.toml @@ -19,7 +19,7 @@ run = "npm --prefix vscode run test:host" [tasks.install-extension] description = "Package and install the extension in local VS Code" -run = ["cargo build -p runa-dap", "npm --prefix vscode run install:local"] +run = "npm --prefix vscode run install:local" [tasks.package-extension] description = "Generate vscode/runa.vsix for manual installation" diff --git a/vscode/README.md b/vscode/README.md index 5118101..57a9cb9 100644 --- a/vscode/README.md +++ b/vscode/README.md @@ -17,12 +17,12 @@ npm --prefix vscode ci mise run install-extension ``` -This builds `runa-dap`, compiles the extension, packages `vscode/runa.vsix`, and +This compiles the extension, packages `vscode/runa.vsix`, and installs it with `code --install-extension ... --force`. Reload the VS Code window after installation. Set `VSCODE_CLI` to a different CLI executable if necessary, for example `VSCODE_CLI=code-insiders mise run install-extension`. -Set `runa.debugAdapterPath` to the absolute path of `target/debug/runa-dap` -to enable debugging (or install `runa-dap` on PATH). Installation does not change +The image server hosts the debugger; no separate adapter binary is required. +Installation does not change your VS Code settings or start an image. The VSIX uses the local extension ID `runa.runa`; packaging is local-only and skips the currently absent license and repository metadata checks, not publication. @@ -35,8 +35,8 @@ mise run package-extension This compiles the extension and writes `vscode/runa.vsix`. Install it using VS Code's **Extensions: Install from VSIX...** command or -`code --install-extension vscode/runa.vsix`. The VSIX does not bundle the Rust -debug adapter; build or install `runa-dap` separately for debugging. +`code --install-extension vscode/runa.vsix`. Debugging connects directly to the +running image server's DAP endpoint. From the repository root: @@ -266,7 +266,7 @@ image session; failed actors are inspected without retrying them. Disconnecting from the image ends its debug session without changing actors. Set `runa.autoAttachDebugger` to `false` to disable automatic attachment on the next connection. Debugger startup errors are reported without dropping the image -connection; check `runa.debugAdapterPath` if the executable cannot be launched. +connection; check `runa.dapAddress` and the server's `--dap-listen` if it cannot connect. With automatic attachment disabled, right-click an actor and choose **Debug Actor**. Alternatively choose **Runa Actor** in Run and Debug and use an attach @@ -282,13 +282,10 @@ configuration (omit `actorRef` to choose an actor): **Debug Actor** does not save open editors. Starting an attach configuration directly through Run and Debug uses VS Code's `debug.saveBeforeStart` setting. -The extension launches the standalone Rust `runa-dap` Debug Adapter Protocol -server over standard input/output. Build it with `cargo build -p runa-dap` -and set `runa.debugAdapterPath` to the executable's absolute path (for example -`/path/to/runa/target/debug/runa-dap`), or install it on PATH with -`cargo install --path crates/runa-dap` from the repository root. -The adapter connects to the selected image over loopback WebSocket; it does not -start an image, spawn an actor, or open a separate debug port. +The extension connects directly to the image server's DAP endpoint using +`runa.dapAddress` (default `127.0.0.1:4711`). The server hosts the Rust DAP library +and dispatches debugger requests through the same image worker as the WebSocket +API. It does not launch a separate adapter executable or spawn actors. An image-wide session discovers one thread per actor incarnation. Removal, restart, or a new loaded implementation retires only that actor's old thread and its inspection handles; the image session and other actors remain usable. @@ -299,12 +296,12 @@ invocation if any. An idle actor can be attached, but has no frames until it starts a handler. An already-failed actor can also be attached without changing its status, incarnation, state or mailbox. -For a DAP client connecting over TCP, start -`runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737`. +To customize the server's DAP listener, start +`cargo run -p runa -- --dap-listen 127.0.0.1:4711`. The optional `RUNA_DAP_LISTEN` environment variable sets the same bind address. -The listener is unauthenticated and restricted to loopback; it serves one -session at a time. Port `0` selects an available port and prints it on stderr. -The extension continues to launch its own stdio adapter by default. +The listener is unauthenticated and restricted to loopback. Port `0` selects an +available port and prints it on stderr. Set `runa.dapAddress` to the resulting +address. The old `runa.debugAdapterPath` setting and standalone binary are removed. - Set gutter breakpoints in each actor's `behavior.rn`, or its immutable loaded source, opened by the command or Call Stack view. The debugger serves the exact compiled text, diff --git a/vscode/integration/dap.test.ts b/vscode/integration/dap.test.ts index 9b9ee3f..9e50522 100644 --- a/vscode/integration/dap.test.ts +++ b/vscode/integration/dap.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; -import { spawn } from "node:child_process"; -import { EventEmitter, once } from "node:events"; -import { resolve } from "node:path"; +import { EventEmitter } from "node:events"; +import { connect } from "node:net"; +import { parseEndpoint } from "../src/client"; import test from "node:test"; import { ImageClient } from "../src/client"; import { parseValue } from "../src/values"; @@ -17,31 +17,20 @@ interface DapMessage { } function startAdapter(address: string) { - const child = spawn( - resolve( - __dirname, - "../../../target/debug", - process.platform === "win32" ? "runa-dap.exe" : "runa-dap", - ), - ["--connect", address], - { stdio: ["pipe", "pipe", "pipe"] }, - ); + const { host, port } = parseEndpoint(address); + const socket = connect({ host, port }); const messages: DapMessage[] = []; const changed = new EventEmitter(); let buffer = Buffer.alloc(0); let sequence = 0; let failure: Error | undefined; - let diagnostics = ""; const fail = (error: Error) => { failure = error; changed.emit("message"); }; - child.on("error", fail); - child.stderr.on("data", (chunk: Buffer) => { - diagnostics += chunk.toString(); - }); - child.on("exit", (code) => fail(new Error(`DAP exited (${code}): ${diagnostics}`))); - child.stdout.on("data", (chunk: Buffer) => { + socket.on("error", fail); + socket.on("close", () => fail(new Error("DAP connection closed"))); + socket.on("data", (chunk: Buffer) => { try { buffer = Buffer.concat([buffer, chunk]); for (;;) { @@ -80,11 +69,7 @@ function startAdapter(address: string) { }; const timer = setTimeout(() => { cleanup(); - reject( - new Error( - `DAP response timed out: ${diagnostics}; ${JSON.stringify(messages.slice(-5))}`, - ), - ); + reject(new Error(`DAP response timed out: ${JSON.stringify(messages.slice(-5))}`)); }, 5000); changed.on("message", check); check(); @@ -95,9 +80,7 @@ function startAdapter(address: string) { async request(command: string, args: unknown = {}): Promise { const seq = ++sequence; const body = Buffer.from(JSON.stringify({ seq, type: "request", command, arguments: args })); - child.stdin.write( - Buffer.concat([Buffer.from(`Content-Length: ${body.length}\r\n\r\n`), body]), - ); + socket.write(Buffer.concat([Buffer.from(`Content-Length: ${body.length}\r\n\r\n`), body])); const response = await wait( (message) => message.type === "response" && message.request_seq === seq, ); @@ -105,10 +88,7 @@ function startAdapter(address: string) { return response.body as T; }, async stop() { - if (child.exitCode !== null || child.signalCode !== null) return; - const exited = once(child, "exit"); - child.kill(); - await exited; + socket.destroy(); }, }; } @@ -121,7 +101,7 @@ test( const client = await ImageClient.connect(image.address, (error) => assert.equal(error, undefined), ); - const adapter = startAdapter(image.address); + const adapter = startAdapter(image.dapAddress); t.after(async () => { try { await adapter.stop(); @@ -190,7 +170,7 @@ test( ); test( - "standalone Rust DAP binds behavior.rn breakpoints to one actor and serves exact loaded code", + "image server DAP binds behavior.rn breakpoints to one actor and serves exact loaded code", { timeout: 20000 }, async (t) => { const image = await startImage(); @@ -210,7 +190,7 @@ test( const actor = await client.spawn(file.path, parseValue("10")); const other = await client.spawn(file.path, parseValue("20")); const uri = `runa-actor://1/${actor}/behavior.rn`; - const adapter = startAdapter(image.address); + const adapter = startAdapter(image.dapAddress); t.after(() => adapter.stop()); await adapter.request("initialize", { linesStartAt1: true, columnsStartAt1: true }); await adapter.request("attach", { actorRef: actor, behaviorSourceUri: uri }); @@ -253,7 +233,7 @@ test( ); test( - "standalone Rust DAP inspects failed actors without retrying their invocation", + "image server DAP inspects failed actors without retrying their invocation", { timeout: 20000 }, async (t) => { const image = await startImage(); @@ -277,7 +257,7 @@ test( await client.tick(1, 1000); const before = await client.inspectActor(actor); assert.equal(before.status.status, "failed"); - const adapter = startAdapter(image.address); + const adapter = startAdapter(image.dapAddress); t.after(() => adapter.stop()); await adapter.request("initialize"); await adapter.request("attach", { actorRef: actor }); diff --git a/vscode/integration/host.ts b/vscode/integration/host.ts index 7eee11a..c56bc4c 100644 --- a/vscode/integration/host.ts +++ b/vscode/integration/host.ts @@ -25,17 +25,7 @@ export async function run(): Promise { assert.ok(extension); await vscode.workspace .getConfiguration("runa") - .update( - "debugAdapterPath", - join( - extension.extensionPath, - "..", - "target", - "debug", - process.platform === "win32" ? "runa-dap.exe" : "runa-dap", - ), - vscode.ConfigurationTarget.Global, - ); + .update("dapAddress", image.dapAddress, vscode.ConfigurationTarget.Global); const contributions = extension.packageJSON.contributes; assert.ok( contributions.viewsContainers.activitybar.some( diff --git a/vscode/integration/server.ts b/vscode/integration/server.ts index 248c49a..9b6677d 100644 --- a/vscode/integration/server.ts +++ b/vscode/integration/server.ts @@ -19,6 +19,7 @@ export const COUNTER_SOURCE = `pub fn handle(ctx, state, sender, message) { export async function startImage(): Promise<{ address: string; + dapAddress: string; stop: () => Promise; diagnostics: () => string; }> { @@ -27,7 +28,7 @@ export async function startImage(): Promise<{ "../../../target/debug", process.platform === "win32" ? "runa.exe" : "runa", ); - const child = spawn(executable, ["--listen", "127.0.0.1:0"], { + const child = spawn(executable, ["--listen", "127.0.0.1:0", "--dap-listen", "127.0.0.1:0"], { env: { ...process.env, RUNA_RESTORE: undefined }, stdio: ["ignore", "ignore", "pipe"], }); @@ -37,9 +38,11 @@ export async function startImage(): Promise<{ }); const stop = () => stopImage(child); let address: string; + let dapAddress = ""; try { address = await new Promise((resolve, reject) => { const lines = createInterface({ input: child.stderr }); + let websocketAddress = ""; let diagnostics = ""; const cleanup = () => { clearTimeout(timer); @@ -61,10 +64,11 @@ export async function startImage(): Promise<{ child.once("exit", exited); lines.on("line", (line) => { diagnostics = (diagnostics + line + "\n").slice(-8192); - const address = line.match(/^runa: listening on (.+)$/)?.[1]; - if (address) { + websocketAddress = line.match(/^runa: listening on (.+)$/)?.[1] ?? websocketAddress; + dapAddress = line.match(/^runa: dap listening on (.+)$/)?.[1] ?? dapAddress; + if (websocketAddress && dapAddress) { cleanup(); - resolve(address); + resolve(websocketAddress); } }); }); @@ -84,7 +88,7 @@ export async function startImage(): Promise<{ await stop(); throw error; } - return { address, stop, diagnostics: () => diagnostics }; + return { address, dapAddress, stop, diagnostics: () => diagnostics }; } async function stopImage(child: ChildProcess): Promise { diff --git a/vscode/package.json b/vscode/package.json index 2fb6908..10b3252 100644 --- a/vscode/package.json +++ b/vscode/package.json @@ -27,8 +27,8 @@ "lint": "tsc -p . --noEmit && prettier --check .", "format": "prettier --write .", "test": "npm run compile && node --test out/test/*.test.js", - "test:integration": "cargo build --manifest-path ../Cargo.toml -p runa -p runa-cli -p runa-dap && npm run compile && node --test out/integration/*.test.js", - "test:host": "cargo build --manifest-path ../Cargo.toml -p runa -p runa-dap && npm run compile && node scripts/test-host.cjs" + "test:integration": "cargo build --manifest-path ../Cargo.toml -p runa -p runa-cli && npm run compile && node --test out/integration/*.test.js", + "test:host": "cargo build --manifest-path ../Cargo.toml -p runa && npm run compile && node scripts/test-host.cjs" }, "dependencies": { "lossless-json": "^4.3.0", @@ -639,11 +639,11 @@ "scope": "machine", "description": "Trusted local image endpoint: numeric IPv4 loopback or [::1]:port. No authentication is available." }, - "runa.debugAdapterPath": { + "runa.dapAddress": { "type": "string", - "default": "runa-dap", + "default": "127.0.0.1:4711", "scope": "machine", - "description": "Path to the standalone Rust runa-dap executable. Build with cargo build -p runa-dap, or install it on PATH." + "description": "Trusted image server's DAP endpoint. Must match runa --dap-listen; numeric loopback address only." }, "runa.autoAttachDebugger": { "type": "boolean", diff --git a/vscode/src/extension.ts b/vscode/src/extension.ts index 588af47..9729f11 100644 --- a/vscode/src/extension.ts +++ b/vscode/src/extension.ts @@ -302,7 +302,9 @@ export function activate(context: vscode.ExtensionContext): { { suppressSaveBeforeStart: true }, ); if (!started) - throw new Error("Could not start the image debugger; check runa.debugAdapterPath"); + throw new Error( + "Could not start the image debugger; check runa.dapAddress and the server's --dap-listen address", + ); } catch (error) { report(error); } @@ -1053,13 +1055,12 @@ export function activate(context: vscode.ExtensionContext): { ) ) throw new Error("An overlapping Runa debug session is already active"); - const executable = vscode.workspace + const address = vscode.workspace .getConfiguration("runa") - .get("debugAdapterPath", "runa-dap"); - if (!executable.trim()) - throw new Error("Set runa.debugAdapterPath to the runa-dap executable"); + .get("dapAddress", "127.0.0.1:4711"); + const { host, port } = parseEndpoint(address); debuggers.set(debugSession.id, debugSession); - return new vscode.DebugAdapterExecutable(executable, ["--connect", connection.endpoint]); + return new vscode.DebugAdapterServer(port, host); }, }), vscode.debug.onDidTerminateDebugSession((session) => debuggers.delete(session.id)),