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)),