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>
This commit is contained in:
Steve Biedermann 2026-11-05 10:21:25 +01:00
parent 8211ef6da4
commit 1e586d41a4
24 changed files with 907 additions and 430 deletions

View File

@ -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. 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 - Add `mise run package-extension` to generate `vscode/runa.vsix` for manual
installation without installing it automatically. installation without installing it automatically.
- Add `runa-dap --listen` / `RUNA_DAP_LISTEN` for loopback TCP DAP sessions, - Add `mise run install-extension` to package the VS Code
retaining stdio as the default transport.
- Add `mise run install-extension` to build the debug adapter, package the VS Code
extension, and install it locally using `code` (or `VSCODE_CLI`). 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 - Expose `behavior.rn` for every actor in the graphical and VS Code trees, with
actor-specific gutter breakpoints and source diagnostics. actor-specific gutter breakpoints and source diagnostics.
- Add an egui/eframe graphical client for browsing and inspecting actors in a - 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 ### 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 - 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 tree and a right-hand actor tree with metadata, state, mailbox, log, and behavior
entries. entries.

4
Cargo.lock generated
View File

@ -2554,6 +2554,7 @@ dependencies = [
"runa-actor", "runa-actor",
"runa-client", "runa-client",
"runa-code", "runa-code",
"runa-dap",
"runa-image-api", "runa-image-api",
"runa-protocol", "runa-protocol",
"serde_json", "serde_json",
@ -2618,9 +2619,6 @@ dependencies = [
name = "runa-dap" name = "runa-dap"
version = "0.0.0" version = "0.0.0"
dependencies = [ dependencies = [
"clap",
"runa",
"runa-client",
"runa-core", "runa-core",
"runa-protocol", "runa-protocol",
"serde", "serde",

View File

@ -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 exits either interface without stopping the image. `cancel` is no longer a
REPL command; the underlying runtime cancellation API remains available. 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 `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 simultaneous clients sharing one live image. The listener rejects non-loopback
addresses and has no authentication; any local process that can connect has 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. 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 The `runa-client` crate connects over loopback
WebSockets and exchange typed protocol requests with request/response correlation. WebSockets and exchange typed protocol requests with request/response correlation.
@ -264,15 +267,15 @@ the runtime contract.
## VS Code client ## VS Code client
To install into your regular VS Code instance, run `npm --prefix vscode ci` 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 followed by `mise run install-extension`, then reload the window. This
Rust debug adapter and packages/installs the extension using the local `code` packages/installs the extension using the local `code`
CLI. Set `VSCODE_CLI` to choose another executable, such as `code-insiders`. CLI. Set `VSCODE_CLI` to choose another executable, such as `code-insiders`.
See [extension setup](vscode/README.md#run-from-source) for debugger configuration. See [extension setup](vscode/README.md#run-from-source) for debugger configuration.
To generate an installable file without installing it, run To generate an installable file without installing it, run
`mise run package-extension`. The output is `vscode/runa.vsix`, ready for `mise run package-extension`. The output is `vscode/runa.vsix`, ready for
VS Code's **Extensions: Install from VSIX...** command. The Rust debug adapter VS Code's **Extensions: Install from VSIX...** command. No separate debug adapter
is built/installed separately. executable is needed.
With Node.js 22+ and npm installed, run `npm --prefix vscode ci`. In VS Code, 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 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. at the next handler boundary without changing shared source, state, or mailbox.
Every actor directory includes `behavior.rn`, even without a temporary override. 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 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 Run and Debug controls. The image server hosts DAP directly at `127.0.0.1:4711`;
`cargo build -p runa-dap` and set `runa.debugAdapterPath` to its absolute path, set `runa.dapAddress` to match a custom server `--dap-listen` address.
or install it on PATH with `cargo install --path crates/runa-dap`. The old standalone `runa-dap` executable and `runa.debugAdapterPath` setting
For TCP DAP clients, use `runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737` have been removed.
(or set `RUNA_DAP_LISTEN`). The DAP listener is loopback-only; omitting the
listen option keeps stdio transport.
**Reload Actor Source** **Reload Actor Source**
clears the override, **Go to Behavior** opens the shared source, and **Update Behavior** 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. 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/ ├── crates/
│ ├── runa/ # Image server, protocol dispatcher, WebSocket transport │ ├── runa/ # Image server, protocol dispatcher, WebSocket transport
│ ├── runa-cli/ # Interactive REPL client │ ├── 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-core/ # Shared stable identities and portable actor values
│ ├── runa-actor/ # Actor value contracts │ ├── runa-actor/ # Actor value contracts
│ ├── runa-code/ # Source identities and compile cache │ ├── runa-code/ # Source identities and compile cache

View File

@ -4,12 +4,7 @@ version = "0.0.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
clap.workspace = true
runa-client = { path = "../runa-client" }
runa-core = { path = "../runa-core" } runa-core = { path = "../runa-core" }
runa-protocol = { path = "../runa-protocol" } runa-protocol = { path = "../runa-protocol" }
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
[dev-dependencies]
runa = { path = "../runa" }

View File

@ -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::cell::RefCell;
use std::collections::{BTreeMap, BTreeSet}; use std::collections::{BTreeMap, BTreeSet};
use std::net::SocketAddr;
use std::rc::Rc; use std::rc::Rc;
use runa_client::WebSocketClient;
use runa_protocol::{ use runa_protocol::{
ActorSnapshotDto, ActorStatusDto, Command, ExecutionInspectionDto, Reply, SourceBreakpoint, ActorSnapshotDto, ActorStatusDto, Command, ExecutionInspectionDto, Reply, SourceBreakpoint,
StackFrame, StepMode, SystemMessageDto, StackFrame, StepMode, SystemMessageDto,
@ -20,36 +19,6 @@ pub trait Connection {
fn request(&mut self, command: Command) -> Result<Reply, String>; fn request(&mut self, command: Command) -> Result<Reply, String>;
} }
pub struct WebSocketConnection {
address: SocketAddr,
client: Option<WebSocketClient>,
}
impl WebSocketConnection {
pub fn new(address: SocketAddr) -> Self {
Self {
address,
client: None,
}
}
}
impl Connection for WebSocketConnection {
fn request(&mut self, command: Command) -> Result<Reply, String> {
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)] #[derive(Deserialize)]
struct Request { struct Request {
seq: u64, seq: u64,

View File

@ -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<SocketAddr>,
}
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(())
}

View File

@ -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<Value> {
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();
}

View File

@ -8,6 +8,7 @@ clap.workspace = true
futures-util = { version = "0.3", features = ["sink"] } futures-util = { version = "0.3", features = ["sink"] }
runa-actor = { path = "../runa-actor" } runa-actor = { path = "../runa-actor" }
runa-code = { path = "../runa-code" } runa-code = { path = "../runa-code" }
runa-dap = { path = "../runa-dap" }
runa-image-api = { path = "../runa-image-api" } runa-image-api = { path = "../runa-image-api" }
runa-protocol = { path = "../runa-protocol" } runa-protocol = { path = "../runa-protocol" }
serde_json = "1" serde_json = "1"

175
crates/runa/src/dap.rs Normal file
View File

@ -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<Dispatch>,
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<Dispatch>,
next_request: u64,
}
impl Connection for WorkerConnection {
fn request(&mut self, command: Command) -> Result<Reply, String> {
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<Instant>,
}
impl Read for DeadlineReader {
fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
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<Option<Json>>;
fn read_frames(stream: TcpStream, sender: mpsc::SyncSender<Frame>) {
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<Dispatch>) -> 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
}

View File

@ -4,9 +4,10 @@ use std::io;
use std::net::{SocketAddr, TcpListener}; use std::net::{SocketAddr, TcpListener};
use std::num::NonZeroUsize; use std::num::NonZeroUsize;
mod dap;
mod network; mod network;
mod storage; mod storage;
pub use network::serve_websocket; pub use network::{serve, serve_websocket};
use runa_actor::{ActorRef, Event, Status, SystemMessage}; use runa_actor::{ActorRef, Event, Status, SystemMessage};
use runa_image_api::ImageApi; use runa_image_api::ImageApi;

View File

@ -3,7 +3,7 @@ use std::net::SocketAddr;
use std::process::ExitCode; use std::process::ExitCode;
use clap::Parser; 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}; use runa_protocol::{Command, PROTOCOL_VERSION, Request};
#[derive(Debug, Parser)] #[derive(Debug, Parser)]
@ -12,6 +12,9 @@ struct Args {
/// Loopback WebSocket listen address (port 0 selects an ephemeral port). /// Loopback WebSocket listen address (port 0 selects an ephemeral port).
#[arg(long, env = "RUNA_LISTEN", default_value = "127.0.0.1:3737")] #[arg(long, env = "RUNA_LISTEN", default_value = "127.0.0.1:3737")]
listen: SocketAddr, 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. /// Restore a server-local image snapshot instead of loading bundled sources.
#[arg(long, env = "RUNA_RESTORE")] #[arg(long, env = "RUNA_RESTORE")]
restore: Option<String>, restore: Option<String>,
@ -42,9 +45,13 @@ fn initial_image(restore: Option<&str>) -> io::Result<Server> {
fn main() -> ExitCode { fn main() -> ExitCode {
let args = Args::parse(); let args = Args::parse();
match bind_loopback_listener(args.listen) // Bind both endpoints and load the image before advertising either listener.
.and_then(|listener| serve_websocket(listener, initial_image(args.restore.as_deref())?)) 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, Ok(()) => ExitCode::SUCCESS,
Err(error) => { Err(error) => {
eprintln!("runa: {error}"); eprintln!("runa: {error}");
@ -65,6 +72,23 @@ mod tests {
.find(|argument| argument.get_id() == "listen") .find(|argument| argument.get_id() == "listen")
.unwrap(); .unwrap();
assert_eq!(listen.get_default_values(), ["127.0.0.1:3737"]); 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::<SocketAddr>().unwrap()
);
assert!(Args::try_parse_from(["runa", "--dap-listen"]).is_err());
assert!(Args::try_parse_from(["runa", "--dap-listen", "invalid"]).is_err());
assert_eq!( assert_eq!(
Args::try_parse_from(["runa", "--listen", "[::1]:4000"]) Args::try_parse_from(["runa", "--listen", "[::1]:4000"])
.unwrap() .unwrap()

View File

@ -1,12 +1,13 @@
use std::io; use std::io;
use std::net::TcpListener; use std::net::TcpListener;
use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use runa_protocol::{ use runa_protocol::{
Command, ImageEvent, Notification, PROTOCOL_VERSION, Request, Response, WEBSOCKET_SUBPROTOCOL, 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::task::JoinSet;
use tokio::time::timeout; use tokio::time::timeout;
use tokio_tungstenite::tungstenite::handshake::server::{ 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::http::{HeaderValue, StatusCode};
use tokio_tungstenite::tungstenite::{Message, protocol::WebSocketConfig}; 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 MAX_NETWORK_CLIENTS: usize = 16;
const WRITE_TIMEOUT: Duration = Duration::from_secs(30); 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<()> { pub fn serve_websocket(listener: TcpListener, server: Server) -> io::Result<()> {
if !listener.local_addr()?.ip().is_loopback() { run_server(listener, None, server)
return Err(io::Error::new( }
io::ErrorKind::InvalidInput,
"network listener must be bound to a loopback address", /// 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<TcpListener>, 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() tokio::runtime::Builder::new_current_thread()
.enable_all() .enable_all()
.build()? .build()?
.block_on(serve(listener, server)) .block_on(run(websocket, dap, server))
} }
struct Dispatch { pub(crate) enum Responder {
request: Request, Async(oneshot::Sender<Response>),
respond: oneshot::Sender<Response>, Blocking(std::sync::mpsc::SyncSender<Response>),
} }
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<TcpListener>,
mut server: Server,
) -> io::Result<()> {
let listener = tokio::net::TcpListener::from_std(listener)?; 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()?); 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::<Dispatch>(MAX_NETWORK_CLIENTS); let (requests, mut pending) = mpsc::channel::<Dispatch>(MAX_NETWORK_CLIENTS);
let (changes, _) = watch::channel(0_u64); let (changes, _) = watch::channel(0_u64);
let mut revision = 0_u64; let mut revision = 0_u64;
let mut clients = JoinSet::new(); let mut clients = JoinSet::new();
let dap_sessions = Arc::new(Semaphore::new(dap::MAX_SESSIONS));
let (logs, _) = broadcast::channel::<runa_protocol::LogEntry>(1024); let (logs, _) = broadcast::channel::<runa_protocol::LogEntry>(1024);
let log_sender = logs.clone(); let log_sender = logs.clone();
server.image.set_log_sink(move |entry| { 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() => { Some(dispatch) = pending.recv() => {
let mutates = !matches!(dispatch.request.command, let mutates = !matches!(dispatch.request.command,
Command::GetBehavior { .. } | Command::GetBehaviors | 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 response = server.dispatch(dispatch.request);
let changed = mutates && response.result.is_ok(); let changed = mutates && response.result.is_ok();
// Dispatch may commit even if the initiating client has disconnected. // 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 { if changed {
revision = revision.checked_add(1) revision = revision.checked_add(1)
.ok_or_else(|| io::Error::other("image notification revisions exhausted"))?; .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 receiver = requested_logs.as_ref().filter(|(_, follow)| *follow).map(|_| logs.subscribe());
let (respond, response) = oneshot::channel(); let (respond, response) = oneshot::channel();
requests.send(Dispatch { request, respond }).await?; requests.send(Dispatch { request, respond: Responder::Async(respond) }).await?;
let response = response.await?; let response = response.await?;
if let (Some((actors, _)), Ok(runa_protocol::Reply::Logs { next_sequence, .. })) = (requested_logs, &response.result) { if let (Some((actors, _)), Ok(runa_protocol::Reply::Logs { next_sequence, .. })) = (requested_logs, &response.result) {
subscription = receiver; subscription = receiver;

View File

@ -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<TcpStream>,
writer: TcpStream,
seq: u64,
pending: Vec<Json>,
}
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<TcpStream> {
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<TcpStream>) -> 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<Dap> = (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}"
);
}
}

View File

@ -49,6 +49,7 @@ fn start_server_with_restore(
) -> (ServerProcess, SocketAddr) { ) -> (ServerProcess, SocketAddr) {
let mut command = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")); let mut command = ProcessCommand::new(env!("CARGO_BIN_EXE_runa"));
command command
.env("RUNA_DAP_LISTEN", "127.0.0.1:0")
.args(args) .args(args)
.env("RUNA_LISTEN", listen_env) .env("RUNA_LISTEN", listen_env)
.env_remove("RUNA_RESTORE"); .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"], vec!["--listen", "0.0.0.0:0"],
] { ] {
let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa"))
.env("RUNA_DAP_LISTEN", "127.0.0.1:0")
.args(args) .args(args)
.env_remove("RUNA_LISTEN") .env_remove("RUNA_LISTEN")
.env_remove("RUNA_RESTORE") .env_remove("RUNA_RESTORE")
@ -99,6 +101,7 @@ fn server_clap_supports_env_overrides_help_and_explicit_errors() {
assert!(!output.stderr.is_empty()); assert!(!output.stderr.is_empty());
} }
let help = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) let help = ProcessCommand::new(env!("CARGO_BIN_EXE_runa"))
.env("RUNA_DAP_LISTEN", "127.0.0.1:0")
.arg("--help") .arg("--help")
.env_remove("RUNA_LISTEN") .env_remove("RUNA_LISTEN")
.env_remove("RUNA_RESTORE") .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")) 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]) .args(["--listen", "127.0.0.1:0", "--restore", file])
.env_remove("RUNA_LISTEN") .env_remove("RUNA_LISTEN")
.env_remove("RUNA_RESTORE") .env_remove("RUNA_RESTORE")
@ -255,6 +259,7 @@ fn startup_restore_failures_exit_without_advertising_a_live_image() {
assert!(!error.contains("listening on"), "{error}"); assert!(!error.contains("listening on"), "{error}");
} }
let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) 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"]) .args(["--listen", "127.0.0.1:0"])
.env("RUNA_RESTORE", missing) .env("RUNA_RESTORE", missing)
.output() .output()
@ -534,6 +539,7 @@ impl Drop for ServerProcess {
#[test] #[test]
fn websocket_server_accepts_local_protocol_requests() { fn websocket_server_accepts_local_protocol_requests() {
let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) 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"]) .args(["--listen", "127.0.0.1:0"])
.env_remove("RUNA_RESTORE") .env_remove("RUNA_RESTORE")
.stdout(Stdio::null()) .stdout(Stdio::null())

View File

@ -240,7 +240,7 @@ unavailable.
## 8. Unified Programmatic Image API ## 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): 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, Await locations represent the last safe snapshot; native/hidden async frames,
locals and post-mortem stacks remain unimplemented. Actor-specific locals and post-mortem stacks remain unimplemented. Actor-specific
implementation-bound line breakpoints and bounded step-into/over/out now preserve active 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` drafts/effects and use exact loaded-source mappings. The image server hosts the
server speaks DAP over stdio and connects to the image over loopback WebSocket. Rust `runa-dap` library as a loopback TCP DAP endpoint, dispatching through the
The VS Code extension launches it as an executable adapter. Every actor exposes 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. `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 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 sources, gutter breakpoints, call stacks, pause/resume, and bounded toolbar

View File

@ -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 strings as step replies. Installation clears breakpoints; restart/cancel retain
them for unchanged loaded code and clear stop reasons. Debugger state is actor 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. 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. DAP gutter breakpoints to these actor-local controls.
System message shape is `{kind: "stop"|"kill"|"restart"|"pause"|"resume"|"reload"}` 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 the next handler boundary. Normal scheduling advances through `tick`; the
explicit debugger step operation advances only its selected paused actor. 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 The image server speaks standard Content-Length-framed DAP over loopback TCP,
on stdin/stdout and connects to an existing trusted loopback image using the defaulting to `127.0.0.1:4711`. Set `runa --dap-listen 127.0.0.1:4712` or
WebSocket operations above. Diagnostics go to stderr. It does not start an `RUNA_DAP_LISTEN` to change the bind address; the CLI flag takes precedence.
image or tick the scheduler implicitly. Stdio mode opens no DAP network listener. Port `0` chooses an available port, reported on stderr as
`runa: dap listening on ...`. Only loopback addresses are accepted because the
To accept DAP over TCP instead of stdio, use API is unauthenticated. DAP requests dispatch through the same single image
`runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737`. worker as WebSocket requests; no second image or adapter process is created.
`RUNA_DAP_LISTEN` also selects the DAP bind address; `--listen` overrides it. Debugger controls do not tick the scheduler implicitly. The standalone binary
Only loopback addresses are accepted because the API is unauthenticated. and stdio DAP transport have been removed.
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.
An attach request supplies `actorRef`, optionally `behaviorSourceUri` (the An attach request supplies `actorRef`, optionally `behaviorSourceUri` (the
actor's `behavior.rn` editor URI) and `workingSourceUri` (its shared image source actor's `behavior.rn` editor URI) and `workingSourceUri` (its shared image source

View File

@ -19,7 +19,7 @@ run = "npm --prefix vscode run test:host"
[tasks.install-extension] [tasks.install-extension]
description = "Package and install the extension in local VS Code" 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] [tasks.package-extension]
description = "Generate vscode/runa.vsix for manual installation" description = "Generate vscode/runa.vsix for manual installation"

View File

@ -17,12 +17,12 @@ npm --prefix vscode ci
mise run install-extension 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 installs it with `code --install-extension ... --force`. Reload the VS Code window
after installation. Set `VSCODE_CLI` to a different CLI executable if necessary, after installation. Set `VSCODE_CLI` to a different CLI executable if necessary,
for example `VSCODE_CLI=code-insiders mise run install-extension`. for example `VSCODE_CLI=code-insiders mise run install-extension`.
Set `runa.debugAdapterPath` to the absolute path of `target/debug/runa-dap` The image server hosts the debugger; no separate adapter binary is required.
to enable debugging (or install `runa-dap` on PATH). Installation does not change Installation does not change
your VS Code settings or start an image. The VSIX uses the local extension ID 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 `runa.runa`; packaging is local-only and skips the currently absent license and
repository metadata checks, not publication. 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 This compiles the extension and writes `vscode/runa.vsix`. Install it using
VS Code's **Extensions: Install from VSIX...** command or VS Code's **Extensions: Install from VSIX...** command or
`code --install-extension vscode/runa.vsix`. The VSIX does not bundle the Rust `code --install-extension vscode/runa.vsix`. Debugging connects directly to the
debug adapter; build or install `runa-dap` separately for debugging. running image server's DAP endpoint.
From the repository root: 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. Disconnecting from the image ends its debug session without changing actors.
Set `runa.autoAttachDebugger` to `false` to disable automatic attachment on the Set `runa.autoAttachDebugger` to `false` to disable automatic attachment on the
next connection. Debugger startup errors are reported without dropping the image 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**. With automatic attachment disabled, right-click an actor and choose **Debug Actor**.
Alternatively choose **Runa Actor** in Run and Debug and use an attach 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 **Debug Actor** does not save open editors. Starting an attach configuration
directly through Run and Debug uses VS Code's `debug.saveBeforeStart` setting. directly through Run and Debug uses VS Code's `debug.saveBeforeStart` setting.
The extension launches the standalone Rust `runa-dap` Debug Adapter Protocol The extension connects directly to the image server's DAP endpoint using
server over standard input/output. Build it with `cargo build -p runa-dap` `runa.dapAddress` (default `127.0.0.1:4711`). The server hosts the Rust DAP library
and set `runa.debugAdapterPath` to the executable's absolute path (for example and dispatches debugger requests through the same image worker as the WebSocket
`/path/to/runa/target/debug/runa-dap`), or install it on PATH with API. It does not launch a separate adapter executable or spawn actors.
`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.
An image-wide session discovers one thread per actor incarnation. Removal, An image-wide session discovers one thread per actor incarnation. Removal,
restart, or a new loaded implementation retires only that actor's old thread 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. 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 starts a handler. An already-failed actor can also be attached without changing
its status, incarnation, state or mailbox. its status, incarnation, state or mailbox.
For a DAP client connecting over TCP, start To customize the server's DAP listener, start
`runa-dap --listen 127.0.0.1:4711 --connect 127.0.0.1:3737`. `cargo run -p runa -- --dap-listen 127.0.0.1:4711`.
The optional `RUNA_DAP_LISTEN` environment variable sets the same bind address. The optional `RUNA_DAP_LISTEN` environment variable sets the same bind address.
The listener is unauthenticated and restricted to loopback; it serves one The listener is unauthenticated and restricted to loopback. Port `0` selects an
session at a time. Port `0` selects an available port and prints it on stderr. available port and prints it on stderr. Set `runa.dapAddress` to the resulting
The extension continues to launch its own stdio adapter by default. 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 - 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, command or Call Stack view. The debugger serves the exact compiled text,

View File

@ -1,7 +1,7 @@
import assert from "node:assert/strict"; import assert from "node:assert/strict";
import { spawn } from "node:child_process"; import { EventEmitter } from "node:events";
import { EventEmitter, once } from "node:events"; import { connect } from "node:net";
import { resolve } from "node:path"; import { parseEndpoint } from "../src/client";
import test from "node:test"; import test from "node:test";
import { ImageClient } from "../src/client"; import { ImageClient } from "../src/client";
import { parseValue } from "../src/values"; import { parseValue } from "../src/values";
@ -17,31 +17,20 @@ interface DapMessage {
} }
function startAdapter(address: string) { function startAdapter(address: string) {
const child = spawn( const { host, port } = parseEndpoint(address);
resolve( const socket = connect({ host, port });
__dirname,
"../../../target/debug",
process.platform === "win32" ? "runa-dap.exe" : "runa-dap",
),
["--connect", address],
{ stdio: ["pipe", "pipe", "pipe"] },
);
const messages: DapMessage[] = []; const messages: DapMessage[] = [];
const changed = new EventEmitter(); const changed = new EventEmitter();
let buffer = Buffer.alloc(0); let buffer = Buffer.alloc(0);
let sequence = 0; let sequence = 0;
let failure: Error | undefined; let failure: Error | undefined;
let diagnostics = "";
const fail = (error: Error) => { const fail = (error: Error) => {
failure = error; failure = error;
changed.emit("message"); changed.emit("message");
}; };
child.on("error", fail); socket.on("error", fail);
child.stderr.on("data", (chunk: Buffer) => { socket.on("close", () => fail(new Error("DAP connection closed")));
diagnostics += chunk.toString(); socket.on("data", (chunk: Buffer) => {
});
child.on("exit", (code) => fail(new Error(`DAP exited (${code}): ${diagnostics}`)));
child.stdout.on("data", (chunk: Buffer) => {
try { try {
buffer = Buffer.concat([buffer, chunk]); buffer = Buffer.concat([buffer, chunk]);
for (;;) { for (;;) {
@ -80,11 +69,7 @@ function startAdapter(address: string) {
}; };
const timer = setTimeout(() => { const timer = setTimeout(() => {
cleanup(); cleanup();
reject( reject(new Error(`DAP response timed out: ${JSON.stringify(messages.slice(-5))}`));
new Error(
`DAP response timed out: ${diagnostics}; ${JSON.stringify(messages.slice(-5))}`,
),
);
}, 5000); }, 5000);
changed.on("message", check); changed.on("message", check);
check(); check();
@ -95,9 +80,7 @@ function startAdapter(address: string) {
async request<T>(command: string, args: unknown = {}): Promise<T> { async request<T>(command: string, args: unknown = {}): Promise<T> {
const seq = ++sequence; const seq = ++sequence;
const body = Buffer.from(JSON.stringify({ seq, type: "request", command, arguments: args })); const body = Buffer.from(JSON.stringify({ seq, type: "request", command, arguments: args }));
child.stdin.write( socket.write(Buffer.concat([Buffer.from(`Content-Length: ${body.length}\r\n\r\n`), body]));
Buffer.concat([Buffer.from(`Content-Length: ${body.length}\r\n\r\n`), body]),
);
const response = await wait( const response = await wait(
(message) => message.type === "response" && message.request_seq === seq, (message) => message.type === "response" && message.request_seq === seq,
); );
@ -105,10 +88,7 @@ function startAdapter(address: string) {
return response.body as T; return response.body as T;
}, },
async stop() { async stop() {
if (child.exitCode !== null || child.signalCode !== null) return; socket.destroy();
const exited = once(child, "exit");
child.kill();
await exited;
}, },
}; };
} }
@ -121,7 +101,7 @@ test(
const client = await ImageClient.connect(image.address, (error) => const client = await ImageClient.connect(image.address, (error) =>
assert.equal(error, undefined), assert.equal(error, undefined),
); );
const adapter = startAdapter(image.address); const adapter = startAdapter(image.dapAddress);
t.after(async () => { t.after(async () => {
try { try {
await adapter.stop(); await adapter.stop();
@ -190,7 +170,7 @@ test(
); );
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 }, { timeout: 20000 },
async (t) => { async (t) => {
const image = await startImage(); const image = await startImage();
@ -210,7 +190,7 @@ test(
const actor = await client.spawn(file.path, parseValue("10")); const actor = await client.spawn(file.path, parseValue("10"));
const other = await client.spawn(file.path, parseValue("20")); const other = await client.spawn(file.path, parseValue("20"));
const uri = `runa-actor://1/${actor}/behavior.rn`; const uri = `runa-actor://1/${actor}/behavior.rn`;
const adapter = startAdapter(image.address); const adapter = startAdapter(image.dapAddress);
t.after(() => adapter.stop()); t.after(() => adapter.stop());
await adapter.request("initialize", { linesStartAt1: true, columnsStartAt1: true }); await adapter.request("initialize", { linesStartAt1: true, columnsStartAt1: true });
await adapter.request("attach", { actorRef: actor, behaviorSourceUri: uri }); await adapter.request("attach", { actorRef: actor, behaviorSourceUri: uri });
@ -253,7 +233,7 @@ test(
); );
test( test(
"standalone Rust DAP inspects failed actors without retrying their invocation", "image server DAP inspects failed actors without retrying their invocation",
{ timeout: 20000 }, { timeout: 20000 },
async (t) => { async (t) => {
const image = await startImage(); const image = await startImage();
@ -277,7 +257,7 @@ test(
await client.tick(1, 1000); await client.tick(1, 1000);
const before = await client.inspectActor(actor); const before = await client.inspectActor(actor);
assert.equal(before.status.status, "failed"); assert.equal(before.status.status, "failed");
const adapter = startAdapter(image.address); const adapter = startAdapter(image.dapAddress);
t.after(() => adapter.stop()); t.after(() => adapter.stop());
await adapter.request("initialize"); await adapter.request("initialize");
await adapter.request("attach", { actorRef: actor }); await adapter.request("attach", { actorRef: actor });

View File

@ -25,17 +25,7 @@ export async function run(): Promise<void> {
assert.ok(extension); assert.ok(extension);
await vscode.workspace await vscode.workspace
.getConfiguration("runa") .getConfiguration("runa")
.update( .update("dapAddress", image.dapAddress, vscode.ConfigurationTarget.Global);
"debugAdapterPath",
join(
extension.extensionPath,
"..",
"target",
"debug",
process.platform === "win32" ? "runa-dap.exe" : "runa-dap",
),
vscode.ConfigurationTarget.Global,
);
const contributions = extension.packageJSON.contributes; const contributions = extension.packageJSON.contributes;
assert.ok( assert.ok(
contributions.viewsContainers.activitybar.some( contributions.viewsContainers.activitybar.some(

View File

@ -19,6 +19,7 @@ export const COUNTER_SOURCE = `pub fn handle(ctx, state, sender, message) {
export async function startImage(): Promise<{ export async function startImage(): Promise<{
address: string; address: string;
dapAddress: string;
stop: () => Promise<void>; stop: () => Promise<void>;
diagnostics: () => string; diagnostics: () => string;
}> { }> {
@ -27,7 +28,7 @@ export async function startImage(): Promise<{
"../../../target/debug", "../../../target/debug",
process.platform === "win32" ? "runa.exe" : "runa", 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 }, env: { ...process.env, RUNA_RESTORE: undefined },
stdio: ["ignore", "ignore", "pipe"], stdio: ["ignore", "ignore", "pipe"],
}); });
@ -37,9 +38,11 @@ export async function startImage(): Promise<{
}); });
const stop = () => stopImage(child); const stop = () => stopImage(child);
let address: string; let address: string;
let dapAddress = "";
try { try {
address = await new Promise<string>((resolve, reject) => { address = await new Promise<string>((resolve, reject) => {
const lines = createInterface({ input: child.stderr }); const lines = createInterface({ input: child.stderr });
let websocketAddress = "";
let diagnostics = ""; let diagnostics = "";
const cleanup = () => { const cleanup = () => {
clearTimeout(timer); clearTimeout(timer);
@ -61,10 +64,11 @@ export async function startImage(): Promise<{
child.once("exit", exited); child.once("exit", exited);
lines.on("line", (line) => { lines.on("line", (line) => {
diagnostics = (diagnostics + line + "\n").slice(-8192); diagnostics = (diagnostics + line + "\n").slice(-8192);
const address = line.match(/^runa: listening on (.+)$/)?.[1]; websocketAddress = line.match(/^runa: listening on (.+)$/)?.[1] ?? websocketAddress;
if (address) { dapAddress = line.match(/^runa: dap listening on (.+)$/)?.[1] ?? dapAddress;
if (websocketAddress && dapAddress) {
cleanup(); cleanup();
resolve(address); resolve(websocketAddress);
} }
}); });
}); });
@ -84,7 +88,7 @@ export async function startImage(): Promise<{
await stop(); await stop();
throw error; throw error;
} }
return { address, stop, diagnostics: () => diagnostics }; return { address, dapAddress, stop, diagnostics: () => diagnostics };
} }
async function stopImage(child: ChildProcess): Promise<void> { async function stopImage(child: ChildProcess): Promise<void> {

View File

@ -27,8 +27,8 @@
"lint": "tsc -p . --noEmit && prettier --check .", "lint": "tsc -p . --noEmit && prettier --check .",
"format": "prettier --write .", "format": "prettier --write .",
"test": "npm run compile && node --test out/test/*.test.js", "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: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 -p runa-dap && npm run compile && node scripts/test-host.cjs" "test:host": "cargo build --manifest-path ../Cargo.toml -p runa && npm run compile && node scripts/test-host.cjs"
}, },
"dependencies": { "dependencies": {
"lossless-json": "^4.3.0", "lossless-json": "^4.3.0",
@ -639,11 +639,11 @@
"scope": "machine", "scope": "machine",
"description": "Trusted local image endpoint: numeric IPv4 loopback or [::1]:port. No authentication is available." "description": "Trusted local image endpoint: numeric IPv4 loopback or [::1]:port. No authentication is available."
}, },
"runa.debugAdapterPath": { "runa.dapAddress": {
"type": "string", "type": "string",
"default": "runa-dap", "default": "127.0.0.1:4711",
"scope": "machine", "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": { "runa.autoAttachDebugger": {
"type": "boolean", "type": "boolean",

View File

@ -302,7 +302,9 @@ export function activate(context: vscode.ExtensionContext): {
{ suppressSaveBeforeStart: true }, { suppressSaveBeforeStart: true },
); );
if (!started) 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) { } catch (error) {
report(error); report(error);
} }
@ -1053,13 +1055,12 @@ export function activate(context: vscode.ExtensionContext): {
) )
) )
throw new Error("An overlapping Runa debug session is already active"); throw new Error("An overlapping Runa debug session is already active");
const executable = vscode.workspace const address = vscode.workspace
.getConfiguration("runa") .getConfiguration("runa")
.get<string>("debugAdapterPath", "runa-dap"); .get<string>("dapAddress", "127.0.0.1:4711");
if (!executable.trim()) const { host, port } = parseEndpoint(address);
throw new Error("Set runa.debugAdapterPath to the runa-dap executable");
debuggers.set(debugSession.id, debugSession); 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)), vscode.debug.onDidTerminateDebugSession((session) => debuggers.delete(session.id)),