From b63666bcc287d4ccf590f64d7e0af52990440da5 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Tue, 3 Nov 2026 15:49:35 +0100 Subject: [PATCH] feat!: rename server and CLI and adopt Clap arguments Rename the server crate and executable to runa and the REPL to runa-cli. Parse executable arguments with Clap derive/env, default the server to loopback port 3737, and support RUNA_LISTEN and RUNA_CONNECT with explicit flag precedence. Remove the stdio transport and child-process client. Make mise start launch the server and mise cli launch the REPL, updating extension integration runners and documentation. BREAKING CHANGE: runa now starts the WebSocket image server rather than the REPL; use runa-cli for the REPL. The runa-server package, stdio transport, LocalClient, ProtocolClient, and serve/serve-network mise tasks are removed. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- AGENTS.md | 12 +- CHANGELOG.md | 13 +- Cargo.lock | 160 ++- Cargo.toml | 5 +- README.md | 37 +- crates/runa-cli/Cargo.toml | 16 + crates/runa-cli/src/main.rs | 1125 ++++++++++++++++ crates/runa-cli/tests/arguments.rs | 55 + crates/runa-client/src/lib.rs | 242 +--- crates/runa-server/Cargo.toml | 17 - crates/runa-server/src/main.rs | 71 - crates/runa/Cargo.toml | 13 +- crates/{runa-server => runa}/src/lib.rs | 130 +- crates/runa/src/main.rs | 1150 +---------------- crates/{runa-server => runa}/src/network.rs | 16 +- .../tests/local_client.rs | 116 +- docs/architecture.md | 50 +- docs/protocol.md | 24 +- docs/runtime.md | 22 +- mise.toml | 12 +- vscode/README.md | 4 +- vscode/integration/repl.test.ts | 16 +- vscode/integration/server.ts | 4 +- vscode/package.json | 4 +- vscode/src/extension.ts | 3 +- 25 files changed, 1585 insertions(+), 1732 deletions(-) create mode 100644 crates/runa-cli/Cargo.toml create mode 100644 crates/runa-cli/src/main.rs create mode 100644 crates/runa-cli/tests/arguments.rs delete mode 100644 crates/runa-server/Cargo.toml delete mode 100644 crates/runa-server/src/main.rs rename crates/{runa-server => runa}/src/lib.rs (78%) rename crates/{runa-server => runa}/src/network.rs (92%) rename crates/{runa-server => runa}/tests/local_client.rs (83%) diff --git a/AGENTS.md b/AGENTS.md index e889ebd..5e5ef16 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -60,7 +60,8 @@ proposed capabilities as plans, not existing behavior. ## Repository structure - `Cargo.toml`: Rust workspace definition. -- `crates/runa/`: current executable crate and entry point. +- `crates/runa/`: image server, protocol dispatch, and loopback WebSocket transport. +- `crates/runa-cli/`: interactive REPL executable. - `crates/runa-core/`: stable identity types shared across layers. - `crates/runa-rune/`: Phase 0 Rune compilation and execution experiments. - `crates/runa-code/`: source identities and in-memory Rune compile cache. @@ -68,8 +69,7 @@ proposed capabilities as plans, not existing behavior. - `crates/runa-runtime/`: single-worker scheduling and in-memory transitions. - `crates/runa-image-api/`: trusted in-process services for operating on an image. - `crates/runa-protocol/`: versioned transport-neutral request, response, and event DTOs. -- `crates/runa-server/`: protocol dispatch, JSON-lines stdio, and loopback WebSockets. -- `crates/runa-client/`: stdio child-process and runa-debug WebSocket clients. +- `crates/runa-client/`: runa-debug WebSocket client. - `runa-std/actors/`: default Rune actor scripts loaded by the local CLI. - `docs/architecture.md`: architecture, roadmap, risks, and acceptance scenario. - `docs/decisions/`: architectural decisions. @@ -90,7 +90,7 @@ mise run format mise run lint mise run test mise run start -mise run serve +mise run cli mise run test-extension-host mise run pre-commit ``` @@ -104,7 +104,9 @@ are required; install the locked extension dependencies with The workspace tests cover execution probes, actor runtime invariants, and the interactive CLI. Extension tests include a real Rust-server WebSocket round trip. -`mise run start` opens the local CLI. The separate `mise run test-extension-host` +`mise run start` starts the image server; `mise run cli` opens the REPL. +Both use Clap derive/env arguments (`RUNA_LISTEN` and `RUNA_CONNECT` respectively). +The separate `mise run test-extension-host` task verifies inspector commands in VS Code and requires a display (or `xvfb-run -a mise run test-extension-host` on headless Linux). diff --git a/CHANGELOG.md b/CHANGELOG.md index 79aef6c..0407993 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,12 @@ All notable changes to this project will be documented in this file. ### Changed +- Renamed the server executable/crate from `runa-server` to `runa`, and the REPL + from `runa` to `runa-cli`. The server now listens on `127.0.0.1:3737` by default. +- Clap derive/env argument parsing for both executables, including generated + help/version, `RUNA_LISTEN` / `RUNA_CONNECT`, and flag-over-environment precedence. +- `mise run start` now starts the server; `mise run cli` launches the REPL. + - Replaced the plain TCP JSON-lines transport with loopback-only WebSockets using the required `runa-debug` subprotocol in the server, REPL, and VS Code. Stdio framing and version 3 request/response shapes are unchanged. @@ -113,4 +119,9 @@ All notable changes to this project will be documented in this file. - Expanded the architecture specification with crate ownership and dependency boundaries, and moved it to `docs/architecture.md`. - Required agents to pass `mise run pre-commit` before every commit and document - user-visible changes in this changelog. \ No newline at end of file + user-visible changes in this changelog. + +### Removed + +- Stdio protocol transport and child-process client; external clients now use + runa-debug WebSockets exclusively. Removed the `serve` and `serve-network` mise tasks. \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 068535c..94f102c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23,6 +23,56 @@ dependencies = [ "memchr", ] +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + [[package]] name = "anyhow" version = "1.0.104" @@ -97,6 +147,46 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" +[[package]] +name = "clap" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9c751b79415d4e559e3d1fcf128e09e720eb673a06d26cf6f392d37d75b66e0" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "clap_lex" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" + [[package]] name = "clipboard-win" version = "5.4.3" @@ -116,6 +206,12 @@ dependencies = [ "unicode-width 0.1.14", ] +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "constant_time_eq" version = "0.4.2" @@ -302,6 +398,12 @@ dependencies = [ "wasip2", ] +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + [[package]] name = "home" version = "0.5.12" @@ -327,6 +429,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itoa" version = "1.0.18" @@ -549,6 +657,12 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "pin-project" version = "1.1.13" @@ -704,13 +818,16 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" name = "runa" version = "0.0.0" dependencies = [ + "clap", + "futures-util", + "runa-actor", "runa-client", - "runa-core", + "runa-code", + "runa-image-api", "runa-protocol", - "runa-server", - "rustyline", "serde_json", - "tungstenite", + "tokio", + "tokio-tungstenite", ] [[package]] @@ -720,6 +837,20 @@ dependencies = [ "runa-core", ] +[[package]] +name = "runa-cli" +version = "0.0.0" +dependencies = [ + "clap", + "runa", + "runa-client", + "runa-core", + "runa-protocol", + "rustyline", + "serde_json", + "tungstenite", +] + [[package]] name = "runa-client" version = "0.0.0" @@ -785,21 +916,6 @@ dependencies = [ "rune", ] -[[package]] -name = "runa-server" -version = "0.0.0" -dependencies = [ - "futures-util", - "runa-actor", - "runa-client", - "runa-code", - "runa-image-api", - "runa-protocol", - "serde_json", - "tokio", - "tokio-tungstenite", -] - [[package]] name = "rune" version = "0.14.2" @@ -1043,6 +1159,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "syn" version = "2.0.119" diff --git a/Cargo.toml b/Cargo.toml index 2fe80b2..215d79d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "3" members = [ "crates/runa", "crates/runa-actor", + "crates/runa-cli", "crates/runa-client", "crates/runa-code", "crates/runa-core", @@ -10,5 +11,7 @@ members = [ "crates/runa-protocol", "crates/runa-rune", "crates/runa-runtime", - "crates/runa-server", ] + +[workspace.dependencies] +clap = { version = "4.5", features = ["derive", "env"] } diff --git a/README.md b/README.md index 3e9f268..663bdca 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,9 @@ documents. The code library now supports an in-memory versioned behavior registry, and the CLI exposes publishing, history, rollback, safe-boundary installation, pause/resume, and cancellation through a trusted in-process image API. A versioned transport-neutral protocol contract is now defined, but no -authentication is implemented. The dispatcher is available through local -JSON-lines stdio and an unauthenticated WebSocket listener restricted to localhost; -both use a bounded request size. The interactive CLI and VS Code connect with +authentication is implemented. The `runa` server provides an unauthenticated +WebSocket listener restricted to localhost, with bounded request sizes. +The `runa-cli` REPL and VS Code connect with the `runa-debug` WebSocket subprotocol, so clients share one server-owned image. Server change notifications automatically refresh VS Code's actor tree and inspectors. @@ -51,33 +51,40 @@ You need a Rust toolchain that supports the workspace's Rust 2024 edition. Start the live image from the repository root: ```sh -mise run serve-network +mise run start ``` In another terminal, connect the REPL: ```sh -mise run start +mise run cli ``` The REPL defaults to `127.0.0.1:3737`. To select another loopback endpoint, -use `cargo run -p runa -- --connect 127.0.0.1:4000` and start the server with +use `cargo run -p runa-cli -- --connect 127.0.0.1:4000` and start the server with the corresponding `--listen` address. Quitting the REPL disconnects only that client; actors and published behaviors survive until the server exits. Reconnect and use `actors` to recover their references. REPL bindings and command history are session-local and are not restored on reconnect. -For a local child-process protocol endpoint, run `mise run serve` and exchange -one JSON request per line on stdin; responses are written to stdout. Request -lines are limited to 1 MiB. +Both executables use Clap for `--help`, `--version`, and argument validation. +`runa` listens on `127.0.0.1:3737` without arguments. Set `RUNA_LISTEN` to change +the server address and `RUNA_CONNECT` to change the REPL address; explicit +`--listen` and `--connect` flags override their respective environment variables. +For example: -For a WebSocket endpoint, run `mise run serve-network`. It listens on +```sh +RUNA_LISTEN=127.0.0.1:4000 mise run start +RUNA_CONNECT=127.0.0.1:4000 mise run cli +``` + +The server is WebSocket-only; 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 `runa-client` crate can launch the stdio process or connect over loopback +The `runa-client` crate connects over loopback WebSockets and exchange typed protocol requests with request/response correlation. Network connection establishment is limited to 5 seconds and response waits/writes to 30 seconds. Transport failures disconnect the client and are never automatically @@ -130,7 +137,7 @@ the runtime contract. 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 -(F5). Start `mise run serve-network`, then use **Runa: Connect to Local Image** +(F5). Start `mise run start`, then use **Runa: Connect to Local Image** in the Extension Development Host. The **Runa Actors** view in Explorer lists actors created through the REPL or another client. Click an actor to inspect it. Server notifications automatically refresh the tree and open snapshots; @@ -174,7 +181,8 @@ use `xvfb-run -a mise run test-extension-host`. . ├── Cargo.toml ├── crates/ -│ ├── runa/ # Main binary +│ ├── runa/ # Image server, protocol dispatcher, WebSocket transport +│ ├── runa-cli/ # Interactive REPL client │ ├── runa-core/ # Shared stable identities and portable actor values │ ├── runa-actor/ # Actor value contracts │ ├── runa-code/ # Source identities and compile cache @@ -182,8 +190,7 @@ use `xvfb-run -a mise run test-extension-host`. │ ├── runa-runtime/ # Single-worker actor prototype │ ├── runa-image-api/ # Trusted in-process image operations │ ├── runa-protocol/ # Versioned transport-neutral wire DTOs -│ ├── runa-server/ # Protocol dispatcher and local JSON-lines transport -│ └── runa-client/ # Stdio child-process and loopback WebSocket clients +│ └── runa-client/ # Loopback WebSocket client ├── docs/ │ ├── architecture.md │ ├── protocol.md diff --git a/crates/runa-cli/Cargo.toml b/crates/runa-cli/Cargo.toml new file mode 100644 index 0000000..3df9206 --- /dev/null +++ b/crates/runa-cli/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "runa-cli" +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" } +rustyline = "15.0" +serde_json = "1" + +[dev-dependencies] +runa = { path = "../runa" } +tungstenite = "0.28" diff --git a/crates/runa-cli/src/main.rs b/crates/runa-cli/src/main.rs new file mode 100644 index 0000000..82c76b5 --- /dev/null +++ b/crates/runa-cli/src/main.rs @@ -0,0 +1,1125 @@ +use std::collections::BTreeMap; +#[cfg(test)] +use std::io::BufRead; +use std::io::{self, Write}; +use std::net::SocketAddr; +use std::num::NonZeroUsize; + +#[cfg(test)] +use runa::Server; +use runa_client::WebSocketClient; +use runa_core::ActorRef; +use runa_protocol::{Command, Reply, Value}; +use rustyline::completion::{Completer, Pair}; +use rustyline::error::ReadlineError; +use rustyline::highlight::Highlighter; +use rustyline::hint::Hinter; +use rustyline::history::DefaultHistory; +use rustyline::validate::Validator; +use rustyline::{Context, Editor, Helper}; + +const MAILBOX_CAPACITY: usize = 64; +const DEFAULT_TICK_BUDGET: usize = 32; +#[cfg(test)] +const COUNTER_BEHAVIOR: &str = "counter"; + +trait Connection { + fn dispatch(&mut self, command: Command) -> Result; +} + +impl Connection for WebSocketClient { + fn dispatch(&mut self, command: Command) -> Result { + self.request(command) + .map_err(|error| error.to_string())? + .result + .map_err(|error| error.message) + } +} + +#[cfg(test)] +impl Connection for Server { + fn dispatch(&mut self, command: Command) -> Result { + Server::dispatch( + self, + runa_protocol::Request { + version: runa_protocol::PROTOCOL_VERSION, + request_id: "runa-cli-test".to_owned(), + command, + }, + ) + .result + .map_err(|error| error.message) + } +} + +#[derive(Default)] +struct ReplState { + bindings: BTreeMap, +} + +#[derive(Default)] +struct ReplHelper { + behaviors: Vec, + bindings: Vec<(String, String)>, +} + +impl ReplHelper { + fn set_context(&mut self, behaviors: Vec, bindings: &BTreeMap) { + self.behaviors = behaviors; + self.bindings = bindings + .iter() + .map(|(name, actor_ref)| (name.clone(), actor_ref.clone())) + .collect(); + } + + fn candidates(&self, line: &str, position: usize) -> Vec { + const COMMANDS: &[&str] = &[ + "actors", + "behaviors", + "cancel", + "events", + "exit", + "help", + "history", + "inspect", + "install", + "let", + "pause", + "publish", + "quit", + "restart", + "resume", + "rollback", + "send", + "spawn", + "stop", + "tick", + ]; + let (start, reference_value) = Self::completion_start(line, position); + if reference_value { + return self + .actor_candidates() + .into_iter() + .filter(|option| option.starts_with(&line[start..position])) + .collect(); + } + let previous: Vec<_> = line[..start].split_whitespace().collect(); + let options: Vec = match previous.as_slice() { + [] => COMMANDS + .iter() + .map(|command| (*command).to_owned()) + .collect(), + ["let", _, "="] => vec!["spawn".to_owned()], + ["spawn"] | ["let", _, "=", "spawn"] => self.behaviors.clone(), + ["let", _, "=", "spawn", _] => Vec::new(), + ["send"] + | ["inspect"] + | ["stop"] + | ["restart"] + | ["pause"] + | ["resume"] + | ["cancel"] + | ["install"] => self.actor_candidates(), + ["install", _] | ["history"] | ["rollback"] | ["publish"] => self.behaviors.clone(), + _ => Vec::new(), + }; + let prefix = &line[start..position]; + options + .into_iter() + .filter(|option| option.starts_with(prefix)) + .collect() + } + + fn completion_start(line: &str, position: usize) -> (usize, bool) { + let before_cursor = &line[..position]; + if let Some(marker) = before_cursor.rfind("\"$actor\"") { + let suffix = &before_cursor[marker + "\"$actor\"".len()..]; + if let Some(value) = suffix.trim_start().strip_prefix(':') { + let value = value.trim_start(); + if let Some(content) = value.strip_prefix('"') + && !content.contains('"') + { + return (position - content.len(), true); + } + } + } + ( + before_cursor + .char_indices() + .rfind(|(_, character)| character.is_whitespace()) + .map_or(0, |(index, character)| index + character.len_utf8()), + false, + ) + } + + fn actor_candidates(&self) -> Vec { + self.bindings + .iter() + .flat_map(|(name, actor_ref)| [name.clone(), format!("@{actor_ref}")]) + .collect() + } +} + +impl Completer for ReplHelper { + type Candidate = Pair; + + fn complete( + &self, + line: &str, + position: usize, + _context: &Context<'_>, + ) -> rustyline::Result<(usize, Vec)> { + let (start, _) = ReplHelper::completion_start(line, position); + let candidates = self + .candidates(line, position) + .into_iter() + .map(|candidate| Pair { + display: candidate.clone(), + replacement: candidate, + }) + .collect(); + Ok((start, candidates)) + } +} + +impl Hinter for ReplHelper { + type Hint = String; +} + +impl Highlighter for ReplHelper {} +impl Validator for ReplHelper {} +impl Helper for ReplHelper {} + +#[derive(Debug, clap::Parser)] +#[command(version, about = "Connect to a localhost Runa image")] +struct Args { + /// Loopback address of the runa-debug WebSocket server. + #[arg(long, env = "RUNA_CONNECT", default_value = "127.0.0.1:3737")] + connect: SocketAddr, +} + +fn main() -> Result<(), Box> { + let address = ::parse().connect; + let mut server = WebSocketClient::connect(address).map_err(|error| { + format!("cannot connect to {address}: {error}; start `mise run start` first") + })?; + + println!("Runa — Rune Actors"); + println!("Connected to {address}. The image remains running when you quit."); + println!("Type `help` for commands. Use Up/Down for history and Tab for completion."); + run_interactive_repl(&mut server)?; + Ok(()) +} + +#[cfg(test)] +fn run_repl( + mut input: R, + mut output: W, + server: &mut dyn Connection, +) -> io::Result<()> { + let mut state = ReplState::default(); + let mut line = String::new(); + loop { + write!(output, "runa> ")?; + output.flush()?; + line.clear(); + if input.read_line(&mut line)? == 0 { + writeln!(output)?; + break; + } + if !execute_line(&line, &mut output, server, &mut state)? { + break; + } + } + Ok(()) +} + +fn run_interactive_repl(server: &mut dyn Connection) -> Result<(), Box> { + let mut editor = Editor::::new()?; + editor.set_helper(Some(ReplHelper::default())); + let mut state = ReplState::default(); + let mut output = io::stdout().lock(); + loop { + let behaviors = match dispatch(server, Command::GetBehaviors)? { + Reply::Behaviors(behaviors) => behaviors + .into_iter() + .map(|behavior| behavior.name) + .collect(), + other => return Err(format!("unexpected protocol response: {other:?}").into()), + }; + editor + .helper_mut() + .expect("REPL helper is installed") + .set_context(behaviors, &state.bindings); + match editor.readline("runa> ") { + Ok(line) => { + if !line.trim().is_empty() { + editor.add_history_entry(line.as_str())?; + } + if !execute_line(&line, &mut output, server, &mut state)? { + break; + } + } + Err(ReadlineError::Interrupted) => writeln!(output, "^C")?, + Err(ReadlineError::Eof) => { + writeln!(output)?; + break; + } + Err(error) => return Err(error.into()), + } + } + Ok(()) +} + +fn execute_line( + line: &str, + output: &mut W, + server: &mut dyn Connection, + state: &mut ReplState, +) -> io::Result { + let words: Vec<_> = line.split_whitespace().collect(); + if words.is_empty() { + return Ok(true); + } + if let ["let", name, "=", "spawn", behavior_name, initial_state @ ..] = words.as_slice() + && !initial_state.is_empty() + { + if !is_binding_name(name) { + writeln!(output, "error: invalid actor binding name `{name}`")?; + return Ok(true); + } + if state.bindings.contains_key(*name) { + writeln!(output, "error: actor binding `{name}` already exists")?; + return Ok(true); + } + match spawn_actor(behavior_name, after_words(line, 5), server, state) { + Ok(actor_ref) => { + writeln!(output, "{name} = @{actor_ref}")?; + state.bindings.insert((*name).to_owned(), actor_ref); + } + Err(error) => writeln!(output, "error: {error}")?, + } + return Ok(true); + } + let command = words[0]; + let args = &words[1..]; + let result = match command { + "help" => print_help(output), + "spawn" => spawn_command(output, line, server, state), + "behaviors" if args.is_empty() => match dispatch(server, Command::GetBehaviors) { + Ok(Reply::Behaviors(behaviors)) => { + for behavior in behaviors { + writeln!( + output, + "{} revision {} ({})", + behavior.name, behavior.revision, behavior.current_implementation_id + )?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "behaviors" => writeln!(output, "error: usage: behaviors"), + "publish" => publish_command(output, args, server), + "history" => history_command(output, args, server), + "rollback" => rollback_command(output, args, server), + "install" => install_command(output, args, server, state), + "actors" if args.is_empty() => match dispatch(server, Command::ListActors) { + Ok(Reply::Actors(actors)) => { + for actor in actors { + writeln!(output, "{actor:#?}")?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "actors" => writeln!(output, "error: usage: actors"), + "send" => match parse_send_args(line, state) { + Ok((actor_ref, message)) => { + match dispatch(server, Command::Send { actor_ref, message }) { + Ok(Reply::Acknowledged) => writeln!(output, "message queued"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + Err(error) => writeln!(output, "error: {error}"), + }, + "tick" => match parse_tick_args(args) { + Ok((count, budget)) => { + match dispatch( + server, + Command::Tick { + ticks: count, + instruction_budget: budget.get(), + }, + ) { + Ok(Reply::TickCompleted { ticks }) => { + writeln!(output, "ran {ticks} scheduler tick(s)") + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + Err(error) => writeln!(output, "error: {error}"), + }, + "inspect" => match parse_actor_argument(args, state) { + Ok(actor_ref) => match dispatch(server, Command::InspectActor { actor_ref }) { + Ok(Reply::Actor(actor)) => writeln!(output, "{actor:#?}"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + Err(error) => writeln!(output, "error: {error}"), + }, + "events" if args.is_empty() => match dispatch(server, Command::GetEvents) { + Ok(Reply::Events(events)) => { + for event in events { + writeln!(output, "{event:#?}")?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "events" => writeln!(output, "error: usage: events"), + "stop" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::StopActor { actor_ref } + }), + "restart" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::RestartActor { actor_ref } + }), + "pause" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::PauseActor { actor_ref } + }), + "resume" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::ResumeActor { actor_ref } + }), + "cancel" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::CancelActor { actor_ref } + }), + "quit" | "exit" if args.is_empty() => return Ok(false), + "quit" | "exit" => writeln!(output, "error: usage: quit"), + _ => writeln!(output, "error: unknown command `{command}`; type `help`"), + }; + result?; + Ok(true) +} + +fn lifecycle_command( + output: &mut W, + args: &[&str], + server: &mut dyn Connection, + state: &ReplState, + operation: impl FnOnce(String) -> Command, +) -> io::Result<()> { + match parse_actor_argument(args, state) { + Ok(actor_ref) => match dispatch(server, operation(actor_ref)) { + Ok(Reply::Acknowledged) => writeln!(output, "ok"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + Err(error) => writeln!(output, "error: {error}"), + } +} + +fn spawn_command( + output: &mut W, + line: &str, + server: &mut dyn Connection, + state: &ReplState, +) -> io::Result<()> { + let args: Vec<_> = line.split_whitespace().collect(); + let [_, behavior_name, initial_state @ ..] = args.as_slice() else { + return writeln!( + output, + "error: usage: spawn " + ); + }; + if initial_state.is_empty() { + return writeln!( + output, + "error: usage: spawn " + ); + } + match spawn_actor(behavior_name, after_words(line, 2), server, state) { + Ok(actor_ref) => writeln!(output, "spawned @{actor_ref} from `{behavior_name}`"), + Err(error) => writeln!(output, "error: {error}"), + } +} + +fn spawn_actor( + behavior_name: &str, + initial_state: &str, + server: &mut dyn Connection, + repl: &ReplState, +) -> Result { + let state = parse_value(initial_state, repl) + .map_err(|error| format!("invalid initial state: {error}"))?; + match dispatch( + server, + Command::Spawn { + behavior_name: behavior_name.to_owned(), + state, + mailbox_capacity: MAILBOX_CAPACITY, + }, + )? { + Reply::ActorSpawned { actor_ref } => Ok(actor_ref), + other => Err(format!("unexpected protocol response: {other:?}")), + } +} + +fn parse_actor_argument(args: &[&str], state: &ReplState) -> Result { + let [argument] = args else { + return Err(format!("expected 1 actor reference, got {}", args.len())); + }; + if let Some(actor_ref) = state.bindings.get(*argument) { + return Ok(actor_ref.clone()); + } + let actor_ref = argument.strip_prefix('@').unwrap_or(argument); + if actor_ref.len() != 21 + || !actor_ref + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return Err(format!( + "unknown actor binding or invalid actor reference `{argument}`" + )); + } + Ok(actor_ref.to_owned()) +} + +fn is_binding_name(name: &str) -> bool { + let mut characters = name.chars(); + matches!(characters.next(), Some(character) if character == '_' || character.is_ascii_alphabetic()) + && characters.all(|character| character == '_' || character.is_ascii_alphanumeric()) +} + +fn publish_command( + output: &mut W, + args: &[&str], + server: &mut dyn Connection, +) -> io::Result<()> { + let [name, expected_revision, source_path] = args else { + return writeln!( + output, + "error: usage: publish " + ); + }; + let expected_revision = if *expected_revision == "new" { + None + } else { + match expected_revision.parse::() { + Ok(revision) => Some(revision), + Err(_) => return writeln!(output, "error: expected `new` or a numeric revision"), + } + }; + let source = match std::fs::read_to_string(source_path) { + Ok(source) => source, + Err(error) => { + return writeln!(output, "error: could not read `{source_path}`: {error}"); + } + }; + let behavior = match dispatch( + server, + Command::PublishBehavior { + name: (*name).to_owned(), + expected_revision, + source, + }, + ) { + Ok(Reply::Behavior(behavior)) => behavior, + Ok(other) => return unexpected_reply(output, other), + Err(error) => return writeln!(output, "error: {error}"), + }; + writeln!( + output, + "published `{}` revision {} ({})", + behavior.name, behavior.revision, behavior.current_implementation_id + ) +} + +fn history_command( + output: &mut W, + args: &[&str], + server: &mut dyn Connection, +) -> io::Result<()> { + let [name] = args else { + return writeln!(output, "error: usage: history "); + }; + let behavior = match dispatch( + server, + Command::GetBehavior { + name: (*name).to_owned(), + }, + ) { + Ok(Reply::Behavior(behavior)) => behavior, + Ok(other) => return unexpected_reply(output, other), + Err(error) => return writeln!(output, "error: {error}"), + }; + for version in behavior.history { + let current = if version.revision == behavior.revision { + " (current)" + } else { + "" + }; + writeln!( + output, + "revision {}: {}{}", + version.revision, version.implementation_id, current + )?; + } + Ok(()) +} + +fn rollback_command( + output: &mut W, + args: &[&str], + server: &mut dyn Connection, +) -> io::Result<()> { + let [name, expected_revision, target_revision] = args else { + return writeln!( + output, + "error: usage: rollback " + ); + }; + let expected_revision = match expected_revision.parse::() { + Ok(revision) => revision, + Err(_) => return writeln!(output, "error: invalid expected revision"), + }; + let target_revision = match target_revision.parse::() { + Ok(revision) => revision, + Err(_) => return writeln!(output, "error: invalid target revision"), + }; + match dispatch( + server, + Command::RollbackBehavior { + name: (*name).to_owned(), + expected_revision, + target_revision, + }, + ) { + Ok(Reply::Behavior(behavior)) => writeln!( + output, + "rolled back `{}` to implementation {} as revision {}", + behavior.name, behavior.current_implementation_id, behavior.revision + ), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } +} + +fn install_command( + output: &mut W, + args: &[&str], + server: &mut dyn Connection, + state: &ReplState, +) -> io::Result<()> { + let [actor_arg, name] = args else { + return writeln!( + output, + "error: usage: install " + ); + }; + let actor_ref = match parse_actor_argument(&[*actor_arg], state) { + Ok(actor_ref) => actor_ref, + Err(error) => return writeln!(output, "error: {error}"), + }; + let implementation = match dispatch( + server, + Command::InstallBehavior { + actor_ref: actor_ref.clone(), + behavior_name: (*name).to_owned(), + }, + ) { + Ok(Reply::BehaviorInstalled { implementation_id }) => implementation_id, + Ok(other) => return unexpected_reply(output, other), + Err(error) => return writeln!(output, "error: {error}"), + }; + match dispatch( + server, + Command::GetBehavior { + name: (*name).to_owned(), + }, + ) { + Ok(Reply::Behavior(behavior)) => { + writeln!( + output, + "installed `{}` revision {} on actor @{} ({implementation})", + behavior.name, behavior.revision, actor_ref + ) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } +} + +fn dispatch(server: &mut dyn Connection, command: Command) -> Result { + server.dispatch(command) +} + +fn unexpected_reply(output: &mut W, reply: Reply) -> io::Result<()> { + writeln!(output, "error: unexpected protocol response: {reply:?}") +} + +fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { + match args { + [count] => Ok(( + count + .parse() + .map_err(|_| format!("invalid tick count: `{count}`"))?, + NonZeroUsize::new(DEFAULT_TICK_BUDGET).expect("budget is nonzero"), + )), + [count, budget] => { + let count = count + .parse() + .map_err(|_| format!("invalid tick count: `{count}`"))?; + let budget = budget + .parse() + .map_err(|_| format!("invalid tick budget: `{budget}`"))?; + let budget = NonZeroUsize::new(budget).ok_or("tick budget must be positive")?; + Ok((count, budget)) + } + _ => Err("usage: tick [budget]".to_owned()), + } +} + +fn parse_send_args(line: &str, state: &ReplState) -> Result<(String, Value), String> { + let actor_arg = line + .split_whitespace() + .nth(1) + .ok_or("usage: send ")?; + let actor_ref = parse_actor_argument(&[actor_arg], state)?; + Ok((actor_ref, parse_value(after_words(line, 2), state)?)) +} + +fn after_words(mut line: &str, count: usize) -> &str { + for _ in 0..count { + line = line.trim_start(); + line = line + .find(char::is_whitespace) + .map_or("", |index| &line[index..]); + } + line.trim() +} + +fn parse_value(input: &str, state: &ReplState) -> Result { + if input.len() > runa_core::MAX_VALUE_BYTES { + return Err("JSON value exceeds input byte limit".to_owned()); + } + fn record( + values: serde_json::Map, + state: &ReplState, + ) -> Result { + values + .into_iter() + .map(|(key, value)| Ok((key, convert(value, state)?))) + .collect::, String>>() + .map(Value::Record) + } + fn convert(value: serde_json::Value, state: &ReplState) -> Result { + Ok(match value { + serde_json::Value::Null => Value::Null, + serde_json::Value::Bool(value) => Value::Bool(value), + serde_json::Value::Number(value) => Value::Integer( + value + .as_i64() + .ok_or("numbers must be signed 64-bit integers")?, + ), + serde_json::Value::String(value) => Value::String(value), + serde_json::Value::Array(values) => Value::List( + values + .into_iter() + .map(|value| convert(value, state)) + .collect::>()?, + ), + serde_json::Value::Object(mut values) + if values.len() == 1 && values.contains_key("$actor") => + { + let serde_json::Value::String(reference) = + values.remove("$actor").expect("key checked") + else { + return Err( + "$actor must contain a binding name or actor reference string".to_owned(), + ); + }; + let reference = parse_actor_argument(&[&reference], state)?; + Value::ActorRef(ActorRef::parse(&reference).expect("reference validated")) + } + serde_json::Value::Object(mut values) + if values.len() == 1 && values.contains_key("$record") => + { + let serde_json::Value::Object(values) = + values.remove("$record").expect("key checked") + else { + return Err("$record must contain a JSON object".to_owned()); + }; + record(values, state)? + } + serde_json::Value::Object(values) => record(values, state)?, + }) + } + let value = + serde_json::from_str(input).map_err(|error| format!("invalid JSON value: {error}"))?; + let value = convert(value, state)?; + value.validate()?; + Ok(value) +} + +fn print_help(output: &mut W) -> io::Result<()> { + writeln!( + output, + "Commands:\n\ + let = spawn \n\ + spawn \n\ + behaviors\n\ + publish \n\ + history \n\ + rollback \n\ + install \n\ + send \n\ + State/messages: JSON values; {{\"$actor\":\"name-or-@ref\"}} embeds an actor reference.\n\ + tick [instruction-budget]\n\ + inspect \n\ + actors\n\ + events\n\ + stop \n\ + restart \n\ + pause \n\ + resume \n\ + cancel \n\ + help\n\ + quit" + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn value_parser_preserves_json_shapes_and_distinguishes_references() { + let reference = ActorRef::new(); + let state = ReplState { + bindings: BTreeMap::from([("receiver".to_owned(), reference.to_string())]), + }; + let input = + r#"{"text": "hello world", "items": [null, true, -42, {"$actor": "receiver"}]}"#; + assert_eq!( + parse_value(input, &state).unwrap(), + Value::Record(BTreeMap::from([ + ("text".to_owned(), Value::String("hello world".to_owned())), + ( + "items".to_owned(), + Value::List(vec![ + Value::Null, + Value::Bool(true), + Value::Integer(-42), + Value::ActorRef(reference) + ]) + ), + ])) + ); + assert_eq!( + parse_value(r#"{"$record": {"$actor": "literal"}}"#, &state).unwrap(), + Value::Record(BTreeMap::from([( + "$actor".to_owned(), + Value::String("literal".to_owned()) + )])) + ); + for input in [ + "1.5", + "9223372036854775808", + r#"{"$actor": "missing"}"#, + r#"{"$actor": 1}"#, + "null trailing", + ] { + assert!(parse_value(input, &state).is_err(), "{input}"); + } + } + + #[test] + fn repl_spawns_and_sends_nested_values_without_losing_whitespace() + -> Result<(), Box> { + let mut server = Server::new(); + dispatch( + &mut server, + Command::PublishBehavior { + name: "echo".to_owned(), + expected_revision: None, + source: "pub fn handle(state, message) { (message, []) }".to_owned(), + }, + )?; + let mut state = ReplState::default(); + let mut output = Vec::new(); + execute_line( + r#"let echo = spawn echo {"text": "hello world", "items": [true, null]}"#, + &mut output, + &mut server, + &mut state, + )?; + let actor_ref = state.bindings.get("echo").ok_or("spawn failed")?.clone(); + let Reply::Actor(snapshot) = dispatch( + &mut server, + Command::InspectActor { + actor_ref: actor_ref.clone(), + }, + )? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!( + snapshot.state, + parse_value(r#"{"text":"hello world","items":[true,null]}"#, &state)? + ); + execute_line( + r#"send echo {"nested": [{"$actor": "echo"}, "two spaces"]}"#, + &mut output, + &mut server, + &mut state, + )?; + execute_line("tick 100", &mut output, &mut server, &mut state)?; + let Reply::Actor(snapshot) = dispatch(&mut server, Command::InspectActor { actor_ref })? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!( + snapshot.state, + parse_value(r#"{"nested": [{"$actor":"echo"}, "two spaces"]}"#, &state)? + ); + assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); + assert!(!String::from_utf8(output)?.contains("error:")); + Ok(()) + } + + #[test] + fn connection_arguments_require_an_explicit_socket_address() { + let command = ::command(); + let connect = command + .get_arguments() + .find(|argument| argument.get_id() == "connect") + .unwrap(); + assert_eq!(connect.get_default_values(), ["127.0.0.1:3737"]); + assert_eq!( + ::try_parse_from(["runa-cli", "--connect", "[::1]:4000"]) + .unwrap() + .connect, + "[::1]:4000".parse::().unwrap() + ); + assert!(::try_parse_from(["runa-cli", "--connect"]).is_err()); + assert!( + ::try_parse_from(["runa-cli", "--connect", "invalid"]).is_err() + ); + } + + #[test] + fn websocket_repl_reconnects_to_the_same_image() -> Result<(), Box> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + let worker = std::thread::spawn(move || { + let mut image = server(); + for _ in 0..2 { + let (stream, _) = listener.accept().unwrap(); + struct DebugProtocol; + impl tungstenite::handshake::server::Callback for DebugProtocol { + fn on_request( + self, + _: &tungstenite::handshake::server::Request, + mut response: tungstenite::handshake::server::Response, + ) -> Result< + tungstenite::handshake::server::Response, + tungstenite::handshake::server::ErrorResponse, + > { + response.headers_mut().insert( + "sec-websocket-protocol", + runa_protocol::WEBSOCKET_SUBPROTOCOL.parse().unwrap(), + ); + Ok(response) + } + } + let mut socket = tungstenite::accept_hdr(stream, DebugProtocol).unwrap(); + while let Ok(message) = socket.read() { + let Some(text) = message.to_text().ok() else { + break; + }; + let request = serde_json::from_str(text).unwrap(); + let response = image.dispatch(request); + socket + .send(tungstenite::Message::Text( + serde_json::to_string(&response).unwrap().into(), + )) + .unwrap(); + } + } + }); + + let mut client = WebSocketClient::connect(address)?; + let mut output = Vec::new(); + run_repl( + io::Cursor::new( + "let actor = spawn counter 5\nsend actor {\"op\":\"add\",\"value\":7}\ntick 100\nquit\n", + ), + &mut output, + &mut client, + )?; + let output = String::from_utf8(output)?; + let actor_ref = output + .lines() + .find_map(|line| line.strip_prefix("runa> actor = @")) + .ok_or("spawn did not return an actor reference")?; + assert_eq!(actor_ref.len(), 21); + drop(client); + + let mut client = WebSocketClient::connect(address)?; + let mut output = Vec::new(); + run_repl( + io::Cursor::new(format!("actors\ninspect @{actor_ref}\nquit\n")), + &mut output, + &mut client, + )?; + let output = String::from_utf8(output)?; + assert!(output.contains(actor_ref)); + assert!(!output.contains("error:")); + let Reply::Actor(snapshot) = dispatch( + &mut client, + Command::InspectActor { + actor_ref: actor_ref.to_owned(), + }, + )? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!(snapshot.state, Value::Integer(12)); + assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); + drop(client); + worker.join().expect("WebSocket image worker completed"); + Ok(()) + } + + fn server() -> Server { + let mut server = Server::new(); + dispatch( + &mut server, + Command::PublishBehavior { + name: COUNTER_BEHAVIOR.to_owned(), + expected_revision: None, + source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), + }, + ) + .expect("default behavior publishes"); + server + } + + #[test] + fn repl_drives_actor_messages_and_inspection() -> Result<(), Box> { + let mut server = server(); + let input = io::Cursor::new( + "let sender = spawn counter 0\nlet receiver = spawn counter 10\nsend sender {\"op\": \"send\", \"value\": 5, \"target\": {\"$actor\": \"receiver\"}}\ntick 100\ninspect sender\ninspect receiver\nevents\nquit\n", + ); + let mut output = Vec::new(); + run_repl(input, &mut output, &mut server)?; + let output = String::from_utf8(output)?; + assert!(output.contains("sender = @")); + assert!(output.contains("receiver = @")); + assert!(!output.contains("error:"), "{output}"); + assert!(output.contains("Committed {")); + assert!(output.contains("actor_ref: \"")); + let Reply::Actors(actors) = dispatch(&mut server, Command::ListActors)? else { + return Err("unexpected actors reply".into()); + }; + assert_eq!(actors.len(), 2); + assert!( + actors + .iter() + .all(|actor| actor.status == runa_protocol::ActorStatusDto::Idle) + ); + assert!(actors.iter().any(|actor| actor.state == Value::Integer(1))); + assert!(actors.iter().any(|actor| actor.state == Value::Integer(15))); + Ok(()) + } + + #[test] + fn repl_reports_bad_input_and_continues() -> Result<(), Box> { + let mut server = server(); + let input = io::Cursor::new( + "spawn 0\nspawn counter nope\nsend 000000000000000000000 {\"op\":\"add\",\"value\":1}\ntick 1 0\nunknown\nquit\n", + ); + let mut output = Vec::new(); + run_repl(input, &mut output, &mut server)?; + let output = String::from_utf8(output)?; + assert!(output.contains("usage: spawn ")); + assert!(output.contains("invalid initial state")); + assert!(output.contains("unknown actor")); + assert!(output.contains("tick budget must be positive")); + assert!(output.contains("unknown command")); + Ok(()) + } + + #[test] + fn repl_manages_named_behavior_revisions_and_installs_them() + -> Result<(), Box> { + let mut server = server(); + let source_path = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../runa-std/actors/counter.rn" + ); + let input = io::Cursor::new(format!( + "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nlet actor = spawn counter 0\ninstall actor counter\nsend actor {{\"op\":\"add\",\"value\":1}}\ntick 1 1\ncancel actor\npause actor\nresume actor\ninspect actor\nevents\nquit\n" + )); + let mut output = Vec::new(); + run_repl(input, &mut output, &mut server)?; + let output = String::from_utf8(output)?; + assert!(output.contains("published `counter` revision 2")); + assert!(output.contains("revision 1:")); + assert!(output.contains("revision 2:")); + assert!(output.contains("rolled back `counter`")); + assert!(output.contains("actor = @")); + assert!(output.contains("installed `counter` revision 3 on actor @")); + assert!(output.contains("Paused {")); + assert!(output.contains("Resumed {")); + assert!(output.contains("InvocationCancelled {")); + let behavior = match dispatch( + &mut server, + Command::GetBehavior { + name: COUNTER_BEHAVIOR.to_owned(), + }, + )? { + Reply::Behavior(behavior) => behavior, + other => return Err(format!("unexpected response: {other:?}").into()), + }; + assert_eq!(behavior.revision, 3); + let actor_ref = match dispatch(&mut server, Command::ListActors)? { + Reply::Actors(actors) => actors[0].actor_ref.clone(), + other => return Err(format!("unexpected response: {other:?}").into()), + }; + let actor = match dispatch(&mut server, Command::InspectActor { actor_ref })? { + Reply::Actor(actor) => actor, + other => return Err(format!("unexpected response: {other:?}").into()), + }; + assert_eq!(actor.implementation_id, behavior.current_implementation_id); + Ok(()) + } + + #[test] + fn completion_offers_commands_behaviors_and_named_actor_refs() { + let mut helper = ReplHelper::default(); + let mut bindings = BTreeMap::new(); + bindings.insert("sender".to_owned(), "xK8_pQ2-Lm5_012345678".to_owned()); + helper.set_context(vec!["counter".to_owned()], &bindings); + + assert_eq!(helper.candidates("sp", 2), vec!["spawn"]); + assert_eq!(helper.candidates("spawn co", 8), vec!["counter"]); + assert_eq!(helper.candidates("inspect se", 10), vec!["sender"]); + let unicode_whitespace = "inspect\u{2003}se"; + assert_eq!( + helper.candidates(unicode_whitespace, unicode_whitespace.len()), + vec!["sender"] + ); + let line = r#"send sender {"target": {"$actor": "se"#; + assert_eq!(helper.candidates(line, line.len()), vec!["sender"]); + assert_eq!( + ReplHelper::completion_start(line, line.len()), + (line.len() - 2, true) + ); + } +} diff --git a/crates/runa-cli/tests/arguments.rs b/crates/runa-cli/tests/arguments.rs new file mode 100644 index 0000000..006243c --- /dev/null +++ b/crates/runa-cli/tests/arguments.rs @@ -0,0 +1,55 @@ +use std::process::Command; + +#[test] +fn cli_help_and_version_are_available_without_connecting() { + let output = Command::new(env!("CARGO_BIN_EXE_runa-cli")) + .arg("--help") + .env_remove("RUNA_CONNECT") + .output() + .unwrap(); + assert!(output.status.success()); + let help = String::from_utf8(output.stdout).unwrap(); + assert!(help.contains("runa-cli")); + assert!(help.contains("--connect")); + assert!(help.contains("RUNA_CONNECT")); + assert!(help.contains("127.0.0.1:3737")); + + let output = Command::new(env!("CARGO_BIN_EXE_runa-cli")) + .arg("--version") + .output() + .unwrap(); + assert!(output.status.success()); + assert_eq!( + String::from_utf8(output.stdout).unwrap().trim(), + "runa-cli 0.0.0" + ); +} + +#[test] +fn clap_rejects_bad_flags_addresses_and_environment_values() { + for args in [ + vec!["--connect"], + vec!["--connect", "invalid"], + vec!["--listen", "127.0.0.1:3737"], + vec!["extra"], + ] { + let output = Command::new(env!("CARGO_BIN_EXE_runa-cli")) + .args(args) + .env_remove("RUNA_CONNECT") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert!(!output.stderr.is_empty()); + } + + let output = Command::new(env!("CARGO_BIN_EXE_runa-cli")) + .env("RUNA_CONNECT", "invalid") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert!( + String::from_utf8(output.stderr) + .unwrap() + .contains("--connect") + ); +} diff --git a/crates/runa-client/src/lib.rs b/crates/runa-client/src/lib.rs index 2b1ff68..d12b821 100644 --- a/crates/runa-client/src/lib.rs +++ b/crates/runa-client/src/lib.rs @@ -1,10 +1,8 @@ -//! Stdio and loopback WebSocket clients for the versioned protocol. +//! Loopback WebSocket client for the versioned protocol. -use std::ffi::OsStr; use std::fmt; -use std::io::{self, BufRead, BufReader, BufWriter, Write}; +use std::io; use std::net::{Shutdown, SocketAddr, TcpStream}; -use std::process::{Child, ChildStdin, ChildStdout, Command, ExitStatus, Stdio}; use std::time::{Duration, Instant}; use runa_protocol::{ @@ -14,18 +12,7 @@ use runa_protocol::{ use tungstenite::client::IntoClientRequest; use tungstenite::{Message, WebSocket, protocol::WebSocketConfig}; -pub const MAX_MESSAGE_LINE_BYTES: usize = 1024 * 1024; - -pub struct LocalClient { - child: Child, - client: Option, BufWriter>>, -} - -pub struct ProtocolClient { - reader: R, - writer: W, - next_request_id: u64, -} +pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024; pub struct WebSocketClient { socket: WebSocket, @@ -54,9 +41,9 @@ impl WebSocketClient { WEBSOCKET_SUBPROTOCOL.parse().map_err(io::Error::other)?, ); let config = WebSocketConfig::default() - .max_message_size(Some(MAX_MESSAGE_LINE_BYTES)) - .max_frame_size(Some(MAX_MESSAGE_LINE_BYTES)) - .max_write_buffer_size(2 * MAX_MESSAGE_LINE_BYTES); + .max_message_size(Some(MAX_MESSAGE_BYTES)) + .max_frame_size(Some(MAX_MESSAGE_BYTES)) + .max_write_buffer_size(2 * MAX_MESSAGE_BYTES); let (socket, response) = tungstenite::client::client_with_config(request, stream, Some(config)) .map_err(io::Error::other)?; @@ -114,7 +101,7 @@ impl WebSocketClient { request_id: request_id.clone(), command, })?; - if encoded.len() > MAX_MESSAGE_LINE_BYTES { + if encoded.len() > MAX_MESSAGE_BYTES { return Err(ClientError::RequestTooLarge); } self.socket @@ -180,7 +167,6 @@ pub enum ClientError { Serialization(serde_json::Error), RequestIdsExhausted, RequestTooLarge, - ResponseTooLarge, UnsupportedVersion(u16), CorrelationMismatch { expected: String, actual: String }, } @@ -193,11 +179,7 @@ impl fmt::Display for ClientError { Self::RequestIdsExhausted => write!(formatter, "protocol request IDs exhausted"), Self::RequestTooLarge => write!( formatter, - "protocol request exceeds {MAX_MESSAGE_LINE_BYTES} bytes" - ), - Self::ResponseTooLarge => write!( - formatter, - "protocol response exceeds {MAX_MESSAGE_LINE_BYTES} bytes" + "protocol request exceeds {MAX_MESSAGE_BYTES} bytes" ), Self::UnsupportedVersion(version) => { write!( @@ -235,144 +217,10 @@ impl From for ClientError { } } -impl LocalClient { - pub fn spawn(program: impl AsRef, args: &[impl AsRef]) -> io::Result { - let mut child = Command::new(program) - .args(args) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .spawn()?; - let stdin = child.stdin.take().ok_or_else(|| { - io::Error::other("child stdin was not available after enabling piped stdin") - })?; - let stdout = child.stdout.take().ok_or_else(|| { - io::Error::other("child stdout was not available after enabling piped stdout") - })?; - - Ok(Self { - child, - client: Some(ProtocolClient::new( - BufReader::new(stdout), - BufWriter::new(stdin), - )), - }) - } - - pub fn request(&mut self, command: ProtocolCommand) -> Result { - self.client - .as_mut() - .ok_or_else(|| { - ClientError::Io(io::Error::new( - io::ErrorKind::BrokenPipe, - "client is shut down", - )) - })? - .request(command) - } - - pub fn shutdown(mut self) -> io::Result { - self.client.take(); - self.child.wait() - } -} - -impl ProtocolClient { - pub fn new(reader: R, writer: W) -> Self { - Self { - reader, - writer, - next_request_id: 0, - } - } - - pub fn request(&mut self, command: ProtocolCommand) -> Result { - self.next_request_id = self - .next_request_id - .checked_add(1) - .ok_or(ClientError::RequestIdsExhausted)?; - let request_id = format!("runa-client-{}", self.next_request_id); - let request = Request { - version: PROTOCOL_VERSION, - request_id: request_id.clone(), - command, - }; - let encoded = serde_json::to_vec(&request)?; - if encoded.len() > MAX_MESSAGE_LINE_BYTES { - return Err(ClientError::RequestTooLarge); - } - - self.writer.write_all(&encoded)?; - self.writer.write_all(b"\n")?; - self.writer.flush()?; - - let mut line = Vec::new(); - let has_line = read_bounded_line(&mut self.reader, &mut line).map_err(|error| { - if error.kind() == io::ErrorKind::InvalidData { - ClientError::ResponseTooLarge - } else { - ClientError::Io(error) - } - })?; - if !has_line { - return Err(ClientError::Io(io::Error::new( - io::ErrorKind::UnexpectedEof, - "server closed the connection before responding", - ))); - } - let response: Response = serde_json::from_slice(&line)?; - if response.version != PROTOCOL_VERSION { - return Err(ClientError::UnsupportedVersion(response.version)); - } - if response.request_id != request_id { - return Err(ClientError::CorrelationMismatch { - expected: request_id, - actual: response.request_id, - }); - } - Ok(response) - } -} - -impl Drop for LocalClient { - fn drop(&mut self) { - if self.child.try_wait().ok().flatten().is_none() { - let _ = self.child.kill(); - let _ = self.child.wait(); - } - } -} - -fn read_bounded_line(reader: &mut R, line: &mut Vec) -> io::Result { - line.clear(); - loop { - let available = reader.fill_buf()?; - if available.is_empty() { - return Ok(!line.is_empty()); - } - - let newline = available.iter().position(|byte| *byte == b'\n'); - let payload_len = newline.unwrap_or(available.len()); - if line.len().saturating_add(payload_len) > MAX_MESSAGE_LINE_BYTES { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("response line exceeds {MAX_MESSAGE_LINE_BYTES} bytes"), - )); - } - - line.extend_from_slice(&available[..payload_len]); - let consumed = payload_len + usize::from(newline.is_some()); - reader.consume(consumed); - if newline.is_some() { - return Ok(true); - } - } -} - #[cfg(test)] mod tests { use super::*; - use runa_protocol::{ProtocolError, Reply}; - use std::io::Cursor; + use runa_protocol::Reply; struct DebugProtocol; impl tungstenite::handshake::server::Callback for DebugProtocol { @@ -487,76 +335,4 @@ mod tests { Err(ClientError::Io(error)) if error.kind() == io::ErrorKind::NotConnected)); peer.join().unwrap(); } - - fn response(request_id: &str) -> Vec { - let response = Response { - version: PROTOCOL_VERSION, - request_id: request_id.to_owned(), - result: Ok(Reply::Acknowledged), - }; - let mut bytes = serde_json::to_vec(&response).unwrap(); - bytes.push(b'\n'); - bytes - } - - #[test] - fn client_writes_correlated_requests_and_validates_responses() { - let mut written = Vec::new(); - let mut client = ProtocolClient::new(Cursor::new(response("runa-client-1")), &mut written); - let result = client.request(ProtocolCommand::GetBehaviors).unwrap(); - assert_eq!(result.request_id, "runa-client-1"); - assert_eq!(result.result, Ok(Reply::Acknowledged)); - - let request: Request = serde_json::from_slice( - written - .strip_suffix(b"\n") - .expect("client writes a newline"), - ) - .unwrap(); - assert_eq!(request.request_id, "runa-client-1"); - assert_eq!(request.command, ProtocolCommand::GetBehaviors); - } - - #[test] - fn client_rejects_mismatched_response_ids_and_versions() { - let mut client = ProtocolClient::new(Cursor::new(response("wrong-id")), Vec::new()); - assert!(matches!( - client.request(ProtocolCommand::GetBehaviors), - Err(ClientError::CorrelationMismatch { .. }) - )); - - let mut unsupported = serde_json::to_vec(&Response { - version: PROTOCOL_VERSION + 1, - request_id: "runa-client-1".to_owned(), - result: Err(ProtocolError::new( - runa_protocol::ErrorCode::UnsupportedVersion, - "unsupported", - )), - }) - .unwrap(); - unsupported.push(b'\n'); - let mut client = ProtocolClient::new(Cursor::new(unsupported), Vec::new()); - assert!(matches!( - client.request(ProtocolCommand::GetBehaviors), - Err(ClientError::UnsupportedVersion(version)) if version == PROTOCOL_VERSION + 1 - )); - } - - #[test] - fn client_enforces_request_and_response_size_limits() { - let mut client = ProtocolClient::new(Cursor::new(response("runa-client-1")), Vec::new()); - let error = client - .request(ProtocolCommand::PublishBehavior { - name: "large".to_owned(), - expected_revision: None, - source: "x".repeat(MAX_MESSAGE_LINE_BYTES), - }) - .unwrap_err(); - assert!(matches!(error, ClientError::RequestTooLarge)); - - let oversized_response = vec![b'x'; MAX_MESSAGE_LINE_BYTES + 1]; - let mut client = ProtocolClient::new(Cursor::new(oversized_response), Vec::new()); - let error = client.request(ProtocolCommand::GetBehaviors).unwrap_err(); - assert!(matches!(error, ClientError::ResponseTooLarge)); - } } diff --git a/crates/runa-server/Cargo.toml b/crates/runa-server/Cargo.toml deleted file mode 100644 index 4ac591a..0000000 --- a/crates/runa-server/Cargo.toml +++ /dev/null @@ -1,17 +0,0 @@ -[package] -name = "runa-server" -version = "0.0.0" -edition = "2024" - -[dependencies] -futures-util = { version = "0.3", features = ["sink"] } -runa-actor = { path = "../runa-actor" } -runa-code = { path = "../runa-code" } -runa-image-api = { path = "../runa-image-api" } -runa-protocol = { path = "../runa-protocol" } -serde_json = "1" -tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } -tokio-tungstenite = "0.28" - -[dev-dependencies] -runa-client = { path = "../runa-client" } diff --git a/crates/runa-server/src/main.rs b/crates/runa-server/src/main.rs deleted file mode 100644 index 2699c66..0000000 --- a/crates/runa-server/src/main.rs +++ /dev/null @@ -1,71 +0,0 @@ -use std::io; -use std::net::SocketAddr; -use std::process::ExitCode; - -use runa_protocol::{Command, PROTOCOL_VERSION, Request}; -use runa_server::{Server, bind_loopback_listener, serve_json_lines, serve_websocket}; - -fn initial_image() -> io::Result { - let mut server = Server::new(); - server - .dispatch(Request { - version: PROTOCOL_VERSION, - request_id: "bootstrap-counter".to_owned(), - command: Command::PublishBehavior { - name: "counter".to_owned(), - expected_revision: None, - source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), - }, - }) - .result - .map_err(|error| io::Error::other(error.message))?; - Ok(server) -} - -fn main() -> ExitCode { - match run() { - Ok(()) => ExitCode::SUCCESS, - Err(error) => { - eprintln!("runa-server: {error}"); - ExitCode::FAILURE - } - } -} - -fn run() -> io::Result<()> { - let mut args = std::env::args().skip(1); - match args.next().as_deref() { - None => serve_stdio(), - Some("--listen") => { - let address = args.next().ok_or_else(|| { - io::Error::new( - io::ErrorKind::InvalidInput, - "usage: runa-server [--listen ]", - ) - })?; - if args.next().is_some() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "usage: runa-server [--listen ]", - )); - } - let address = address.parse::().map_err(|error| { - io::Error::new( - io::ErrorKind::InvalidInput, - format!("invalid listener address: {error}"), - ) - })?; - serve_websocket(bind_loopback_listener(address)?, initial_image()?) - } - Some(_) => Err(io::Error::new( - io::ErrorKind::InvalidInput, - "usage: runa-server [--listen ]", - )), - } -} - -fn serve_stdio() -> io::Result<()> { - let stdin = io::stdin(); - let stdout = io::stdout(); - serve_json_lines(&mut Server::new(), stdin.lock(), stdout.lock()) -} diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index 7fd9898..8d4ba06 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -4,12 +4,15 @@ version = "0.0.0" edition = "2024" [dependencies] -runa-client = { path = "../runa-client" } -runa-core = { path = "../runa-core" } +clap.workspace = true +futures-util = { version = "0.3", features = ["sink"] } +runa-actor = { path = "../runa-actor" } +runa-code = { path = "../runa-code" } +runa-image-api = { path = "../runa-image-api" } runa-protocol = { path = "../runa-protocol" } -rustyline = "15.0" serde_json = "1" +tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } +tokio-tungstenite = "0.28" [dev-dependencies] -runa-server = { path = "../runa-server" } -tungstenite = "0.28" +runa-client = { path = "../runa-client" } diff --git a/crates/runa-server/src/lib.rs b/crates/runa/src/lib.rs similarity index 78% rename from crates/runa-server/src/lib.rs rename to crates/runa/src/lib.rs index 46934c3..7031677 100644 --- a/crates/runa-server/src/lib.rs +++ b/crates/runa/src/lib.rs @@ -1,6 +1,6 @@ //! Transport-neutral dispatch from protocol requests to image API operations. -use std::io::{self, BufRead, Write}; +use std::io; use std::net::{SocketAddr, TcpListener}; use std::num::NonZeroUsize; @@ -14,24 +14,12 @@ use runa_protocol::{ EventDto, ExecutionSnapshotDto, PROTOCOL_VERSION, ProtocolError, Reply, Request, Response, }; -pub const MAX_REQUEST_LINE_BYTES: usize = 1024 * 1024; +pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024; pub struct Server { image: ImageApi, } -/// Serve newline-delimited protocol requests over an already-connected stream. -/// -/// Each valid request line receives one JSON response line. Malformed or -/// oversized input terminates the stream with `InvalidData`. -pub fn serve_json_lines( - server: &mut Server, - reader: R, - writer: W, -) -> io::Result<()> { - serve_json_lines_with(reader, writer, |request| Ok(server.dispatch(request))) -} - /// Bind the protocol listener to an IPv4 or IPv6 loopback address only. pub fn bind_loopback_listener(address: SocketAddr) -> io::Result { if !address.ip().is_loopback() { @@ -44,54 +32,6 @@ pub fn bind_loopback_listener(address: SocketAddr) -> io::Result { TcpListener::bind(address) } -fn invalid_data(message: String) -> io::Error { - io::Error::new(io::ErrorKind::InvalidData, message) -} - -fn serve_json_lines_with(mut reader: R, mut writer: W, mut dispatch: F) -> io::Result<()> -where - R: BufRead, - W: Write, - F: FnMut(Request) -> io::Result, -{ - let mut line = Vec::new(); - while read_bounded_line(&mut reader, &mut line)? { - let request: Request = serde_json::from_slice(&line) - .map_err(|error| invalid_data(format!("invalid protocol request JSON: {error}")))?; - let response = dispatch(request)?; - serde_json::to_writer(&mut writer, &response).map_err(io::Error::other)?; - writer.write_all(b"\n")?; - writer.flush()?; - } - Ok(()) -} - -fn read_bounded_line(reader: &mut R, line: &mut Vec) -> io::Result { - line.clear(); - loop { - let available = reader.fill_buf()?; - if available.is_empty() { - return Ok(!line.is_empty()); - } - - let newline = available.iter().position(|byte| *byte == b'\n'); - let payload_len = newline.unwrap_or(available.len()); - if line.len().saturating_add(payload_len) > MAX_REQUEST_LINE_BYTES { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("request line exceeds {MAX_REQUEST_LINE_BYTES} bytes"), - )); - } - - line.extend_from_slice(&available[..payload_len]); - let consumed = payload_len + usize::from(newline.is_some()); - reader.consume(consumed); - if newline.is_some() { - return Ok(true); - } - } -} - impl Default for Server { fn default() -> Self { Self::new() @@ -390,7 +330,6 @@ mod tests { } use runa_protocol::{Command, PROTOCOL_VERSION}; - use std::io::Cursor; fn request(command: Command) -> Request { Request { @@ -532,71 +471,6 @@ mod tests { )); } - #[test] - fn json_lines_transport_dispatches_multiple_requests_and_accepts_final_line_without_newline() { - let publish_request = request(Command::PublishBehavior { - name: "counter".to_owned(), - expected_revision: None, - source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), - }); - let mut spawn_request = request(Command::Spawn { - behavior_name: "counter".to_owned(), - state: Value::Integer(4), - mailbox_capacity: 2, - }); - spawn_request.request_id = "spawn-request".to_owned(); - let mut list_request = request(Command::ListActors); - list_request.request_id = "list-request".to_owned(); - let mut input = [publish_request, spawn_request] - .iter() - .map(serde_json::to_string) - .collect::, _>>() - .unwrap() - .join("\n"); - input.push('\n'); - input.push_str(&serde_json::to_string(&list_request).unwrap()); - - let mut output = Vec::new(); - serve_json_lines(&mut Server::new(), Cursor::new(input), &mut output).unwrap(); - - let responses = output - .split(|byte| *byte == b'\n') - .filter(|line| !line.is_empty()) - .map(serde_json::from_slice::) - .collect::, _>>() - .unwrap(); - assert_eq!(responses.len(), 3); - assert_eq!( - responses - .iter() - .map(|response| response.request_id.as_str()) - .collect::>(), - vec!["test-request", "spawn-request", "list-request"] - ); - assert!(matches!(responses[0].result, Ok(Reply::Behavior(_)))); - assert!(matches!( - &responses[1].result, - Ok(Reply::ActorSpawned { actor_ref }) if actor_ref.len() == 21 - )); - assert!(matches!( - &responses[2].result, - Ok(Reply::Actors(actors)) if actors.len() == 1 && actors[0].state == Value::Integer(4) - )); - } - - #[test] - fn json_lines_transport_rejects_malformed_and_oversized_input() { - let mut server = Server::new(); - let malformed = - serve_json_lines(&mut server, Cursor::new(b"{not-json}\n"), Vec::new()).unwrap_err(); - assert_eq!(malformed.kind(), io::ErrorKind::InvalidData); - - let mut oversized = vec![b'a'; MAX_REQUEST_LINE_BYTES + 1]; - oversized.push(b'\n'); - let error = serve_json_lines(&mut server, Cursor::new(oversized), Vec::new()).unwrap_err(); - assert_eq!(error.kind(), io::ErrorKind::InvalidData); - } - #[test] fn loopback_listener_rejects_non_loopback_addresses() { let error = bind_loopback_listener("0.0.0.0:0".parse().unwrap()).unwrap_err(); diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index 63b1077..125fe6d 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -1,799 +1,47 @@ -use std::collections::BTreeMap; -#[cfg(test)] -use std::io::BufRead; -use std::io::{self, Write}; +use std::io; use std::net::SocketAddr; -use std::num::NonZeroUsize; +use std::process::ExitCode; -use runa_client::WebSocketClient; -use runa_core::ActorRef; -use runa_protocol::{Command, Reply, Value}; -#[cfg(test)] -use runa_server::Server; -use rustyline::completion::{Completer, Pair}; -use rustyline::error::ReadlineError; -use rustyline::highlight::Highlighter; -use rustyline::hint::Hinter; -use rustyline::history::DefaultHistory; -use rustyline::validate::Validator; -use rustyline::{Context, Editor, Helper}; +use clap::Parser; +use runa::{Server, bind_loopback_listener, serve_websocket}; +use runa_protocol::{Command, PROTOCOL_VERSION, Request}; -const MAILBOX_CAPACITY: usize = 64; -const DEFAULT_TICK_BUDGET: usize = 32; -#[cfg(test)] -const COUNTER_BEHAVIOR: &str = "counter"; - -trait Connection { - fn dispatch(&mut self, command: Command) -> Result; +#[derive(Debug, Parser)] +#[command(version, about = "Run a localhost Rune actor image")] +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, } -impl Connection for WebSocketClient { - fn dispatch(&mut self, command: Command) -> Result { - self.request(command) - .map_err(|error| error.to_string())? - .result - .map_err(|error| error.message) - } -} - -#[cfg(test)] -impl Connection for Server { - fn dispatch(&mut self, command: Command) -> Result { - Server::dispatch( - self, - runa_protocol::Request { - version: runa_protocol::PROTOCOL_VERSION, - request_id: "runa-cli-test".to_owned(), - command, +fn initial_image() -> io::Result { + let mut server = Server::new(); + server + .dispatch(Request { + version: PROTOCOL_VERSION, + request_id: "bootstrap-counter".to_owned(), + command: Command::PublishBehavior { + name: "counter".to_owned(), + expected_revision: None, + source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), }, - ) - .result - .map_err(|error| error.message) - } -} - -#[derive(Default)] -struct ReplState { - bindings: BTreeMap, -} - -#[derive(Default)] -struct ReplHelper { - behaviors: Vec, - bindings: Vec<(String, String)>, -} - -impl ReplHelper { - fn set_context(&mut self, behaviors: Vec, bindings: &BTreeMap) { - self.behaviors = behaviors; - self.bindings = bindings - .iter() - .map(|(name, actor_ref)| (name.clone(), actor_ref.clone())) - .collect(); - } - - fn candidates(&self, line: &str, position: usize) -> Vec { - const COMMANDS: &[&str] = &[ - "actors", - "behaviors", - "cancel", - "events", - "exit", - "help", - "history", - "inspect", - "install", - "let", - "pause", - "publish", - "quit", - "restart", - "resume", - "rollback", - "send", - "spawn", - "stop", - "tick", - ]; - let (start, reference_value) = Self::completion_start(line, position); - if reference_value { - return self - .actor_candidates() - .into_iter() - .filter(|option| option.starts_with(&line[start..position])) - .collect(); - } - let previous: Vec<_> = line[..start].split_whitespace().collect(); - let options: Vec = match previous.as_slice() { - [] => COMMANDS - .iter() - .map(|command| (*command).to_owned()) - .collect(), - ["let", _, "="] => vec!["spawn".to_owned()], - ["spawn"] | ["let", _, "=", "spawn"] => self.behaviors.clone(), - ["let", _, "=", "spawn", _] => Vec::new(), - ["send"] - | ["inspect"] - | ["stop"] - | ["restart"] - | ["pause"] - | ["resume"] - | ["cancel"] - | ["install"] => self.actor_candidates(), - ["install", _] | ["history"] | ["rollback"] | ["publish"] => self.behaviors.clone(), - _ => Vec::new(), - }; - let prefix = &line[start..position]; - options - .into_iter() - .filter(|option| option.starts_with(prefix)) - .collect() - } - - fn completion_start(line: &str, position: usize) -> (usize, bool) { - let before_cursor = &line[..position]; - if let Some(marker) = before_cursor.rfind("\"$actor\"") { - let suffix = &before_cursor[marker + "\"$actor\"".len()..]; - if let Some(value) = suffix.trim_start().strip_prefix(':') { - let value = value.trim_start(); - if let Some(content) = value.strip_prefix('"') - && !content.contains('"') - { - return (position - content.len(), true); - } - } - } - ( - before_cursor - .char_indices() - .rfind(|(_, character)| character.is_whitespace()) - .map_or(0, |(index, character)| index + character.len_utf8()), - false, - ) - } - - fn actor_candidates(&self) -> Vec { - self.bindings - .iter() - .flat_map(|(name, actor_ref)| [name.clone(), format!("@{actor_ref}")]) - .collect() - } -} - -impl Completer for ReplHelper { - type Candidate = Pair; - - fn complete( - &self, - line: &str, - position: usize, - _context: &Context<'_>, - ) -> rustyline::Result<(usize, Vec)> { - let (start, _) = ReplHelper::completion_start(line, position); - let candidates = self - .candidates(line, position) - .into_iter() - .map(|candidate| Pair { - display: candidate.clone(), - replacement: candidate, - }) - .collect(); - Ok((start, candidates)) - } -} - -impl Hinter for ReplHelper { - type Hint = String; -} - -impl Highlighter for ReplHelper {} -impl Validator for ReplHelper {} -impl Helper for ReplHelper {} - -fn main() -> Result<(), Box> { - let address = parse_address(&std::env::args().skip(1).collect::>())?; - let mut server = WebSocketClient::connect(address).map_err(|error| { - format!("cannot connect to {address}: {error}; start `mise run serve-network` first") - })?; - - println!("Runa — Rune Actors"); - println!("Connected to {address}. The image remains running when you quit."); - println!("Type `help` for commands. Use Up/Down for history and Tab for completion."); - run_interactive_repl(&mut server)?; - Ok(()) -} - -fn parse_address(args: &[String]) -> Result { - let address = match args { - [] => "127.0.0.1:3737", - [flag, address] if flag == "--connect" => address, - _ => return Err("usage: runa [--connect ]".to_owned()), - }; - address - .parse() - .map_err(|error| format!("invalid server address: {error}")) -} - -#[cfg(test)] -fn run_repl( - mut input: R, - mut output: W, - server: &mut dyn Connection, -) -> io::Result<()> { - let mut state = ReplState::default(); - let mut line = String::new(); - loop { - write!(output, "runa> ")?; - output.flush()?; - line.clear(); - if input.read_line(&mut line)? == 0 { - writeln!(output)?; - break; - } - if !execute_line(&line, &mut output, server, &mut state)? { - break; - } - } - Ok(()) -} - -fn run_interactive_repl(server: &mut dyn Connection) -> Result<(), Box> { - let mut editor = Editor::::new()?; - editor.set_helper(Some(ReplHelper::default())); - let mut state = ReplState::default(); - let mut output = io::stdout().lock(); - loop { - let behaviors = match dispatch(server, Command::GetBehaviors)? { - Reply::Behaviors(behaviors) => behaviors - .into_iter() - .map(|behavior| behavior.name) - .collect(), - other => return Err(format!("unexpected protocol response: {other:?}").into()), - }; - editor - .helper_mut() - .expect("REPL helper is installed") - .set_context(behaviors, &state.bindings); - match editor.readline("runa> ") { - Ok(line) => { - if !line.trim().is_empty() { - editor.add_history_entry(line.as_str())?; - } - if !execute_line(&line, &mut output, server, &mut state)? { - break; - } - } - Err(ReadlineError::Interrupted) => writeln!(output, "^C")?, - Err(ReadlineError::Eof) => { - writeln!(output)?; - break; - } - Err(error) => return Err(error.into()), - } - } - Ok(()) -} - -fn execute_line( - line: &str, - output: &mut W, - server: &mut dyn Connection, - state: &mut ReplState, -) -> io::Result { - let words: Vec<_> = line.split_whitespace().collect(); - if words.is_empty() { - return Ok(true); - } - if let ["let", name, "=", "spawn", behavior_name, initial_state @ ..] = words.as_slice() - && !initial_state.is_empty() - { - if !is_binding_name(name) { - writeln!(output, "error: invalid actor binding name `{name}`")?; - return Ok(true); - } - if state.bindings.contains_key(*name) { - writeln!(output, "error: actor binding `{name}` already exists")?; - return Ok(true); - } - match spawn_actor(behavior_name, after_words(line, 5), server, state) { - Ok(actor_ref) => { - writeln!(output, "{name} = @{actor_ref}")?; - state.bindings.insert((*name).to_owned(), actor_ref); - } - Err(error) => writeln!(output, "error: {error}")?, - } - return Ok(true); - } - let command = words[0]; - let args = &words[1..]; - let result = match command { - "help" => print_help(output), - "spawn" => spawn_command(output, line, server, state), - "behaviors" if args.is_empty() => match dispatch(server, Command::GetBehaviors) { - Ok(Reply::Behaviors(behaviors)) => { - for behavior in behaviors { - writeln!( - output, - "{} revision {} ({})", - behavior.name, behavior.revision, behavior.current_implementation_id - )?; - } - Ok(()) - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "behaviors" => writeln!(output, "error: usage: behaviors"), - "publish" => publish_command(output, args, server), - "history" => history_command(output, args, server), - "rollback" => rollback_command(output, args, server), - "install" => install_command(output, args, server, state), - "actors" if args.is_empty() => match dispatch(server, Command::ListActors) { - Ok(Reply::Actors(actors)) => { - for actor in actors { - writeln!(output, "{actor:#?}")?; - } - Ok(()) - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "actors" => writeln!(output, "error: usage: actors"), - "send" => match parse_send_args(line, state) { - Ok((actor_ref, message)) => { - match dispatch(server, Command::Send { actor_ref, message }) { - Ok(Reply::Acknowledged) => writeln!(output, "message queued"), - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - } - } - Err(error) => writeln!(output, "error: {error}"), - }, - "tick" => match parse_tick_args(args) { - Ok((count, budget)) => { - match dispatch( - server, - Command::Tick { - ticks: count, - instruction_budget: budget.get(), - }, - ) { - Ok(Reply::TickCompleted { ticks }) => { - writeln!(output, "ran {ticks} scheduler tick(s)") - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - } - } - Err(error) => writeln!(output, "error: {error}"), - }, - "inspect" => match parse_actor_argument(args, state) { - Ok(actor_ref) => match dispatch(server, Command::InspectActor { actor_ref }) { - Ok(Reply::Actor(actor)) => writeln!(output, "{actor:#?}"), - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - Err(error) => writeln!(output, "error: {error}"), - }, - "events" if args.is_empty() => match dispatch(server, Command::GetEvents) { - Ok(Reply::Events(events)) => { - for event in events { - writeln!(output, "{event:#?}")?; - } - Ok(()) - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "events" => writeln!(output, "error: usage: events"), - "stop" => lifecycle_command(output, args, server, state, |actor_ref| { - Command::StopActor { actor_ref } - }), - "restart" => lifecycle_command(output, args, server, state, |actor_ref| { - Command::RestartActor { actor_ref } - }), - "pause" => lifecycle_command(output, args, server, state, |actor_ref| { - Command::PauseActor { actor_ref } - }), - "resume" => lifecycle_command(output, args, server, state, |actor_ref| { - Command::ResumeActor { actor_ref } - }), - "cancel" => lifecycle_command(output, args, server, state, |actor_ref| { - Command::CancelActor { actor_ref } - }), - "quit" | "exit" if args.is_empty() => return Ok(false), - "quit" | "exit" => writeln!(output, "error: usage: quit"), - _ => writeln!(output, "error: unknown command `{command}`; type `help`"), - }; - result?; - Ok(true) -} - -fn lifecycle_command( - output: &mut W, - args: &[&str], - server: &mut dyn Connection, - state: &ReplState, - operation: impl FnOnce(String) -> Command, -) -> io::Result<()> { - match parse_actor_argument(args, state) { - Ok(actor_ref) => match dispatch(server, operation(actor_ref)) { - Ok(Reply::Acknowledged) => writeln!(output, "ok"), - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - Err(error) => writeln!(output, "error: {error}"), - } -} - -fn spawn_command( - output: &mut W, - line: &str, - server: &mut dyn Connection, - state: &ReplState, -) -> io::Result<()> { - let args: Vec<_> = line.split_whitespace().collect(); - let [_, behavior_name, initial_state @ ..] = args.as_slice() else { - return writeln!( - output, - "error: usage: spawn " - ); - }; - if initial_state.is_empty() { - return writeln!( - output, - "error: usage: spawn " - ); - } - match spawn_actor(behavior_name, after_words(line, 2), server, state) { - Ok(actor_ref) => writeln!(output, "spawned @{actor_ref} from `{behavior_name}`"), - Err(error) => writeln!(output, "error: {error}"), - } -} - -fn spawn_actor( - behavior_name: &str, - initial_state: &str, - server: &mut dyn Connection, - repl: &ReplState, -) -> Result { - let state = parse_value(initial_state, repl) - .map_err(|error| format!("invalid initial state: {error}"))?; - match dispatch( - server, - Command::Spawn { - behavior_name: behavior_name.to_owned(), - state, - mailbox_capacity: MAILBOX_CAPACITY, - }, - )? { - Reply::ActorSpawned { actor_ref } => Ok(actor_ref), - other => Err(format!("unexpected protocol response: {other:?}")), - } -} - -fn parse_actor_argument(args: &[&str], state: &ReplState) -> Result { - let [argument] = args else { - return Err(format!("expected 1 actor reference, got {}", args.len())); - }; - if let Some(actor_ref) = state.bindings.get(*argument) { - return Ok(actor_ref.clone()); - } - let actor_ref = argument.strip_prefix('@').unwrap_or(argument); - if actor_ref.len() != 21 - || !actor_ref - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) - { - return Err(format!( - "unknown actor binding or invalid actor reference `{argument}`" - )); - } - Ok(actor_ref.to_owned()) -} - -fn is_binding_name(name: &str) -> bool { - let mut characters = name.chars(); - matches!(characters.next(), Some(character) if character == '_' || character.is_ascii_alphabetic()) - && characters.all(|character| character == '_' || character.is_ascii_alphanumeric()) -} - -fn publish_command( - output: &mut W, - args: &[&str], - server: &mut dyn Connection, -) -> io::Result<()> { - let [name, expected_revision, source_path] = args else { - return writeln!( - output, - "error: usage: publish " - ); - }; - let expected_revision = if *expected_revision == "new" { - None - } else { - match expected_revision.parse::() { - Ok(revision) => Some(revision), - Err(_) => return writeln!(output, "error: expected `new` or a numeric revision"), - } - }; - let source = match std::fs::read_to_string(source_path) { - Ok(source) => source, - Err(error) => { - return writeln!(output, "error: could not read `{source_path}`: {error}"); - } - }; - let behavior = match dispatch( - server, - Command::PublishBehavior { - name: (*name).to_owned(), - expected_revision, - source, - }, - ) { - Ok(Reply::Behavior(behavior)) => behavior, - Ok(other) => return unexpected_reply(output, other), - Err(error) => return writeln!(output, "error: {error}"), - }; - writeln!( - output, - "published `{}` revision {} ({})", - behavior.name, behavior.revision, behavior.current_implementation_id - ) -} - -fn history_command( - output: &mut W, - args: &[&str], - server: &mut dyn Connection, -) -> io::Result<()> { - let [name] = args else { - return writeln!(output, "error: usage: history "); - }; - let behavior = match dispatch( - server, - Command::GetBehavior { - name: (*name).to_owned(), - }, - ) { - Ok(Reply::Behavior(behavior)) => behavior, - Ok(other) => return unexpected_reply(output, other), - Err(error) => return writeln!(output, "error: {error}"), - }; - for version in behavior.history { - let current = if version.revision == behavior.revision { - " (current)" - } else { - "" - }; - writeln!( - output, - "revision {}: {}{}", - version.revision, version.implementation_id, current - )?; - } - Ok(()) -} - -fn rollback_command( - output: &mut W, - args: &[&str], - server: &mut dyn Connection, -) -> io::Result<()> { - let [name, expected_revision, target_revision] = args else { - return writeln!( - output, - "error: usage: rollback " - ); - }; - let expected_revision = match expected_revision.parse::() { - Ok(revision) => revision, - Err(_) => return writeln!(output, "error: invalid expected revision"), - }; - let target_revision = match target_revision.parse::() { - Ok(revision) => revision, - Err(_) => return writeln!(output, "error: invalid target revision"), - }; - match dispatch( - server, - Command::RollbackBehavior { - name: (*name).to_owned(), - expected_revision, - target_revision, - }, - ) { - Ok(Reply::Behavior(behavior)) => writeln!( - output, - "rolled back `{}` to implementation {} as revision {}", - behavior.name, behavior.current_implementation_id, behavior.revision - ), - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - } -} - -fn install_command( - output: &mut W, - args: &[&str], - server: &mut dyn Connection, - state: &ReplState, -) -> io::Result<()> { - let [actor_arg, name] = args else { - return writeln!( - output, - "error: usage: install " - ); - }; - let actor_ref = match parse_actor_argument(&[*actor_arg], state) { - Ok(actor_ref) => actor_ref, - Err(error) => return writeln!(output, "error: {error}"), - }; - let implementation = match dispatch( - server, - Command::InstallBehavior { - actor_ref: actor_ref.clone(), - behavior_name: (*name).to_owned(), - }, - ) { - Ok(Reply::BehaviorInstalled { implementation_id }) => implementation_id, - Ok(other) => return unexpected_reply(output, other), - Err(error) => return writeln!(output, "error: {error}"), - }; - match dispatch( - server, - Command::GetBehavior { - name: (*name).to_owned(), - }, - ) { - Ok(Reply::Behavior(behavior)) => { - writeln!( - output, - "installed `{}` revision {} on actor @{} ({implementation})", - behavior.name, behavior.revision, actor_ref - ) - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), - } -} - -fn dispatch(server: &mut dyn Connection, command: Command) -> Result { - server.dispatch(command) -} - -fn unexpected_reply(output: &mut W, reply: Reply) -> io::Result<()> { - writeln!(output, "error: unexpected protocol response: {reply:?}") -} - -fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { - match args { - [count] => Ok(( - count - .parse() - .map_err(|_| format!("invalid tick count: `{count}`"))?, - NonZeroUsize::new(DEFAULT_TICK_BUDGET).expect("budget is nonzero"), - )), - [count, budget] => { - let count = count - .parse() - .map_err(|_| format!("invalid tick count: `{count}`"))?; - let budget = budget - .parse() - .map_err(|_| format!("invalid tick budget: `{budget}`"))?; - let budget = NonZeroUsize::new(budget).ok_or("tick budget must be positive")?; - Ok((count, budget)) - } - _ => Err("usage: tick [budget]".to_owned()), - } -} - -fn parse_send_args(line: &str, state: &ReplState) -> Result<(String, Value), String> { - let actor_arg = line - .split_whitespace() - .nth(1) - .ok_or("usage: send ")?; - let actor_ref = parse_actor_argument(&[actor_arg], state)?; - Ok((actor_ref, parse_value(after_words(line, 2), state)?)) -} - -fn after_words(mut line: &str, count: usize) -> &str { - for _ in 0..count { - line = line.trim_start(); - line = line - .find(char::is_whitespace) - .map_or("", |index| &line[index..]); - } - line.trim() -} - -fn parse_value(input: &str, state: &ReplState) -> Result { - if input.len() > runa_core::MAX_VALUE_BYTES { - return Err("JSON value exceeds input byte limit".to_owned()); - } - fn record( - values: serde_json::Map, - state: &ReplState, - ) -> Result { - values - .into_iter() - .map(|(key, value)| Ok((key, convert(value, state)?))) - .collect::, String>>() - .map(Value::Record) - } - fn convert(value: serde_json::Value, state: &ReplState) -> Result { - Ok(match value { - serde_json::Value::Null => Value::Null, - serde_json::Value::Bool(value) => Value::Bool(value), - serde_json::Value::Number(value) => Value::Integer( - value - .as_i64() - .ok_or("numbers must be signed 64-bit integers")?, - ), - serde_json::Value::String(value) => Value::String(value), - serde_json::Value::Array(values) => Value::List( - values - .into_iter() - .map(|value| convert(value, state)) - .collect::>()?, - ), - serde_json::Value::Object(mut values) - if values.len() == 1 && values.contains_key("$actor") => - { - let serde_json::Value::String(reference) = - values.remove("$actor").expect("key checked") - else { - return Err( - "$actor must contain a binding name or actor reference string".to_owned(), - ); - }; - let reference = parse_actor_argument(&[&reference], state)?; - Value::ActorRef(ActorRef::parse(&reference).expect("reference validated")) - } - serde_json::Value::Object(mut values) - if values.len() == 1 && values.contains_key("$record") => - { - let serde_json::Value::Object(values) = - values.remove("$record").expect("key checked") - else { - return Err("$record must contain a JSON object".to_owned()); - }; - record(values, state)? - } - serde_json::Value::Object(values) => record(values, state)?, }) - } - let value = - serde_json::from_str(input).map_err(|error| format!("invalid JSON value: {error}"))?; - let value = convert(value, state)?; - value.validate()?; - Ok(value) + .result + .map_err(|error| io::Error::other(error.message))?; + Ok(server) } -fn print_help(output: &mut W) -> io::Result<()> { - writeln!( - output, - "Commands:\n\ - let = spawn \n\ - spawn \n\ - behaviors\n\ - publish \n\ - history \n\ - rollback \n\ - install \n\ - send \n\ - State/messages: JSON values; {{\"$actor\":\"name-or-@ref\"}} embeds an actor reference.\n\ - tick [instruction-budget]\n\ - inspect \n\ - actors\n\ - events\n\ - stop \n\ - restart \n\ - pause \n\ - resume \n\ - cancel \n\ - help\n\ - quit" - ) +fn main() -> ExitCode { + let args = Args::parse(); + match bind_loopback_listener(args.listen) + .and_then(|listener| serve_websocket(listener, initial_image()?)) + { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("runa: {error}"); + ExitCode::FAILURE + } + } } #[cfg(test)] @@ -801,322 +49,22 @@ mod tests { use super::*; #[test] - fn value_parser_preserves_json_shapes_and_distinguishes_references() { - let reference = ActorRef::new(); - let state = ReplState { - bindings: BTreeMap::from([("receiver".to_owned(), reference.to_string())]), - }; - let input = - r#"{"text": "hello world", "items": [null, true, -42, {"$actor": "receiver"}]}"#; + fn clap_parses_default_explicit_and_invalid_listeners() { + let command = ::command(); + let listen = command + .get_arguments() + .find(|argument| argument.get_id() == "listen") + .unwrap(); + assert_eq!(listen.get_default_values(), ["127.0.0.1:3737"]); assert_eq!( - parse_value(input, &state).unwrap(), - Value::Record(BTreeMap::from([ - ("text".to_owned(), Value::String("hello world".to_owned())), - ( - "items".to_owned(), - Value::List(vec![ - Value::Null, - Value::Bool(true), - Value::Integer(-42), - Value::ActorRef(reference) - ]) - ), - ])) - ); - assert_eq!( - parse_value(r#"{"$record": {"$actor": "literal"}}"#, &state).unwrap(), - Value::Record(BTreeMap::from([( - "$actor".to_owned(), - Value::String("literal".to_owned()) - )])) - ); - for input in [ - "1.5", - "9223372036854775808", - r#"{"$actor": "missing"}"#, - r#"{"$actor": 1}"#, - "null trailing", - ] { - assert!(parse_value(input, &state).is_err(), "{input}"); - } - } - - #[test] - fn repl_spawns_and_sends_nested_values_without_losing_whitespace() - -> Result<(), Box> { - let mut server = Server::new(); - dispatch( - &mut server, - Command::PublishBehavior { - name: "echo".to_owned(), - expected_revision: None, - source: "pub fn handle(state, message) { (message, []) }".to_owned(), - }, - )?; - let mut state = ReplState::default(); - let mut output = Vec::new(); - execute_line( - r#"let echo = spawn echo {"text": "hello world", "items": [true, null]}"#, - &mut output, - &mut server, - &mut state, - )?; - let actor_ref = state.bindings.get("echo").ok_or("spawn failed")?.clone(); - let Reply::Actor(snapshot) = dispatch( - &mut server, - Command::InspectActor { - actor_ref: actor_ref.clone(), - }, - )? - else { - return Err("unexpected inspect reply".into()); - }; - assert_eq!( - snapshot.state, - parse_value(r#"{"text":"hello world","items":[true,null]}"#, &state)? - ); - execute_line( - r#"send echo {"nested": [{"$actor": "echo"}, "two spaces"]}"#, - &mut output, - &mut server, - &mut state, - )?; - execute_line("tick 100", &mut output, &mut server, &mut state)?; - let Reply::Actor(snapshot) = dispatch(&mut server, Command::InspectActor { actor_ref })? - else { - return Err("unexpected inspect reply".into()); - }; - assert_eq!( - snapshot.state, - parse_value(r#"{"nested": [{"$actor":"echo"}, "two spaces"]}"#, &state)? - ); - assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); - assert!(!String::from_utf8(output)?.contains("error:")); - Ok(()) - } - - #[test] - fn connection_arguments_require_an_explicit_socket_address() { - assert_eq!( - parse_address(&[]).unwrap(), - "127.0.0.1:3737".parse::().unwrap() - ); - assert_eq!( - parse_address(&["--connect".to_owned(), "[::1]:4000".to_owned()]).unwrap(), + Args::try_parse_from(["runa", "--listen", "[::1]:4000"]) + .unwrap() + .listen, "[::1]:4000".parse::().unwrap() ); - assert!(parse_address(&["--connect".to_owned()]).is_err()); - assert!(parse_address(&["--connect".to_owned(), "invalid".to_owned()]).is_err()); - } - - #[test] - fn websocket_repl_reconnects_to_the_same_image() -> Result<(), Box> { - let listener = std::net::TcpListener::bind("127.0.0.1:0")?; - let address = listener.local_addr()?; - let worker = std::thread::spawn(move || { - let mut image = server(); - for _ in 0..2 { - let (stream, _) = listener.accept().unwrap(); - struct DebugProtocol; - impl tungstenite::handshake::server::Callback for DebugProtocol { - fn on_request( - self, - _: &tungstenite::handshake::server::Request, - mut response: tungstenite::handshake::server::Response, - ) -> Result< - tungstenite::handshake::server::Response, - tungstenite::handshake::server::ErrorResponse, - > { - response.headers_mut().insert( - "sec-websocket-protocol", - runa_protocol::WEBSOCKET_SUBPROTOCOL.parse().unwrap(), - ); - Ok(response) - } - } - let mut socket = tungstenite::accept_hdr(stream, DebugProtocol).unwrap(); - while let Ok(message) = socket.read() { - let Some(text) = message.to_text().ok() else { - break; - }; - let request = serde_json::from_str(text).unwrap(); - let response = image.dispatch(request); - socket - .send(tungstenite::Message::Text( - serde_json::to_string(&response).unwrap().into(), - )) - .unwrap(); - } - } - }); - - let mut client = WebSocketClient::connect(address)?; - let mut output = Vec::new(); - run_repl( - io::Cursor::new( - "let actor = spawn counter 5\nsend actor {\"op\":\"add\",\"value\":7}\ntick 100\nquit\n", - ), - &mut output, - &mut client, - )?; - let output = String::from_utf8(output)?; - let actor_ref = output - .lines() - .find_map(|line| line.strip_prefix("runa> actor = @")) - .ok_or("spawn did not return an actor reference")?; - assert_eq!(actor_ref.len(), 21); - drop(client); - - let mut client = WebSocketClient::connect(address)?; - let mut output = Vec::new(); - run_repl( - io::Cursor::new(format!("actors\ninspect @{actor_ref}\nquit\n")), - &mut output, - &mut client, - )?; - let output = String::from_utf8(output)?; - assert!(output.contains(actor_ref)); - assert!(!output.contains("error:")); - let Reply::Actor(snapshot) = dispatch( - &mut client, - Command::InspectActor { - actor_ref: actor_ref.to_owned(), - }, - )? - else { - return Err("unexpected inspect reply".into()); - }; - assert_eq!(snapshot.state, Value::Integer(12)); - assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); - drop(client); - worker.join().expect("WebSocket image worker completed"); - Ok(()) - } - - fn server() -> Server { - let mut server = Server::new(); - dispatch( - &mut server, - Command::PublishBehavior { - name: COUNTER_BEHAVIOR.to_owned(), - expected_revision: None, - source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), - }, - ) - .expect("default behavior publishes"); - server - } - - #[test] - fn repl_drives_actor_messages_and_inspection() -> Result<(), Box> { - let mut server = server(); - let input = io::Cursor::new( - "let sender = spawn counter 0\nlet receiver = spawn counter 10\nsend sender {\"op\": \"send\", \"value\": 5, \"target\": {\"$actor\": \"receiver\"}}\ntick 100\ninspect sender\ninspect receiver\nevents\nquit\n", - ); - let mut output = Vec::new(); - run_repl(input, &mut output, &mut server)?; - let output = String::from_utf8(output)?; - assert!(output.contains("sender = @")); - assert!(output.contains("receiver = @")); - assert!(!output.contains("error:"), "{output}"); - assert!(output.contains("Committed {")); - assert!(output.contains("actor_ref: \"")); - let Reply::Actors(actors) = dispatch(&mut server, Command::ListActors)? else { - return Err("unexpected actors reply".into()); - }; - assert_eq!(actors.len(), 2); - assert!( - actors - .iter() - .all(|actor| actor.status == runa_protocol::ActorStatusDto::Idle) - ); - assert!(actors.iter().any(|actor| actor.state == Value::Integer(1))); - assert!(actors.iter().any(|actor| actor.state == Value::Integer(15))); - Ok(()) - } - - #[test] - fn repl_reports_bad_input_and_continues() -> Result<(), Box> { - let mut server = server(); - let input = io::Cursor::new( - "spawn 0\nspawn counter nope\nsend 000000000000000000000 {\"op\":\"add\",\"value\":1}\ntick 1 0\nunknown\nquit\n", - ); - let mut output = Vec::new(); - run_repl(input, &mut output, &mut server)?; - let output = String::from_utf8(output)?; - assert!(output.contains("usage: spawn ")); - assert!(output.contains("invalid initial state")); - assert!(output.contains("unknown actor")); - assert!(output.contains("tick budget must be positive")); - assert!(output.contains("unknown command")); - Ok(()) - } - - #[test] - fn repl_manages_named_behavior_revisions_and_installs_them() - -> Result<(), Box> { - let mut server = server(); - let source_path = concat!( - env!("CARGO_MANIFEST_DIR"), - "/../../runa-std/actors/counter.rn" - ); - let input = io::Cursor::new(format!( - "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nlet actor = spawn counter 0\ninstall actor counter\nsend actor {{\"op\":\"add\",\"value\":1}}\ntick 1 1\ncancel actor\npause actor\nresume actor\ninspect actor\nevents\nquit\n" - )); - let mut output = Vec::new(); - run_repl(input, &mut output, &mut server)?; - let output = String::from_utf8(output)?; - assert!(output.contains("published `counter` revision 2")); - assert!(output.contains("revision 1:")); - assert!(output.contains("revision 2:")); - assert!(output.contains("rolled back `counter`")); - assert!(output.contains("actor = @")); - assert!(output.contains("installed `counter` revision 3 on actor @")); - assert!(output.contains("Paused {")); - assert!(output.contains("Resumed {")); - assert!(output.contains("InvocationCancelled {")); - let behavior = match dispatch( - &mut server, - Command::GetBehavior { - name: COUNTER_BEHAVIOR.to_owned(), - }, - )? { - Reply::Behavior(behavior) => behavior, - other => return Err(format!("unexpected response: {other:?}").into()), - }; - assert_eq!(behavior.revision, 3); - let actor_ref = match dispatch(&mut server, Command::ListActors)? { - Reply::Actors(actors) => actors[0].actor_ref.clone(), - other => return Err(format!("unexpected response: {other:?}").into()), - }; - let actor = match dispatch(&mut server, Command::InspectActor { actor_ref })? { - Reply::Actor(actor) => actor, - other => return Err(format!("unexpected response: {other:?}").into()), - }; - assert_eq!(actor.implementation_id, behavior.current_implementation_id); - Ok(()) - } - - #[test] - fn completion_offers_commands_behaviors_and_named_actor_refs() { - let mut helper = ReplHelper::default(); - let mut bindings = BTreeMap::new(); - bindings.insert("sender".to_owned(), "xK8_pQ2-Lm5_012345678".to_owned()); - helper.set_context(vec!["counter".to_owned()], &bindings); - - assert_eq!(helper.candidates("sp", 2), vec!["spawn"]); - assert_eq!(helper.candidates("spawn co", 8), vec!["counter"]); - assert_eq!(helper.candidates("inspect se", 10), vec!["sender"]); - let unicode_whitespace = "inspect\u{2003}se"; - assert_eq!( - helper.candidates(unicode_whitespace, unicode_whitespace.len()), - vec!["sender"] - ); - let line = r#"send sender {"target": {"$actor": "se"#; - assert_eq!(helper.candidates(line, line.len()), vec!["sender"]); - assert_eq!( - ReplHelper::completion_start(line, line.len()), - (line.len() - 2, true) - ); + assert!(Args::try_parse_from(["runa", "--listen"]).is_err()); + assert!(Args::try_parse_from(["runa", "--listen", "invalid"]).is_err()); + assert!(Args::try_parse_from(["runa", "--stdio"]).is_err()); + assert!(Args::try_parse_from(["runa", "extra"]).is_err()); } } diff --git a/crates/runa-server/src/network.rs b/crates/runa/src/network.rs similarity index 92% rename from crates/runa-server/src/network.rs rename to crates/runa/src/network.rs index fc6157b..cc6694e 100644 --- a/crates/runa-server/src/network.rs +++ b/crates/runa/src/network.rs @@ -15,7 +15,7 @@ use tokio_tungstenite::tungstenite::handshake::server::{ use tokio_tungstenite::tungstenite::http::{HeaderValue, StatusCode}; use tokio_tungstenite::tungstenite::{Message, protocol::WebSocketConfig}; -use crate::{MAX_REQUEST_LINE_BYTES, Server}; +use crate::{MAX_MESSAGE_BYTES, Server}; const MAX_NETWORK_CLIENTS: usize = 16; const WRITE_TIMEOUT: Duration = Duration::from_secs(30); @@ -42,7 +42,7 @@ struct Dispatch { async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { let listener = tokio::net::TcpListener::from_std(listener)?; - eprintln!("runa-server: listening on {}", listener.local_addr()?); + eprintln!("runa: listening on {}", listener.local_addr()?); let (requests, mut pending) = mpsc::channel::(MAX_NETWORK_CLIENTS); let (changes, _) = watch::channel(0_u64); let mut revision = 0_u64; @@ -52,14 +52,14 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { accepted = listener.accept() => { let (stream, address) = accepted?; if clients.len() >= MAX_NETWORK_CLIENTS { - eprintln!("runa-server: rejecting {address}; client limit reached"); + eprintln!("runa: rejecting {address}; client limit reached"); continue; } let requests = requests.clone(); let changes = changes.subscribe(); clients.spawn(async move { if let Err(error) = client(stream, requests, changes).await { - eprintln!("runa-server: client {address}: {error}"); + eprintln!("runa: client {address}: {error}"); } }); } @@ -125,9 +125,9 @@ async fn client( mut changes: watch::Receiver, ) -> Result<(), Box> { let config = WebSocketConfig::default() - .max_message_size(Some(MAX_REQUEST_LINE_BYTES)) - .max_frame_size(Some(MAX_REQUEST_LINE_BYTES)) - .max_write_buffer_size(2 * MAX_REQUEST_LINE_BYTES); + .max_message_size(Some(MAX_MESSAGE_BYTES)) + .max_frame_size(Some(MAX_MESSAGE_BYTES)) + .max_write_buffer_size(2 * MAX_MESSAGE_BYTES); let mut socket = timeout( Duration::from_secs(5), tokio_tungstenite::accept_hdr_async_with_config(stream, DebugProtocol, Some(config)), @@ -142,7 +142,7 @@ async fn client( let (respond, response) = oneshot::channel(); requests.send(Dispatch { request, respond }).await?; let encoded = serde_json::to_string(&response.await?)?; - if encoded.len() > MAX_REQUEST_LINE_BYTES { + if encoded.len() > MAX_MESSAGE_BYTES { return Err("WebSocket response exceeds 1 MiB".into()); } timeout(WRITE_TIMEOUT, socket.send(Message::Text(encoded.into()))).await??; diff --git a/crates/runa-server/tests/local_client.rs b/crates/runa/tests/local_client.rs similarity index 83% rename from crates/runa-server/tests/local_client.rs rename to crates/runa/tests/local_client.rs index 82e2aee..cd20f3d 100644 --- a/crates/runa-server/tests/local_client.rs +++ b/crates/runa/tests/local_client.rs @@ -3,13 +3,19 @@ use std::io::{BufRead, BufReader}; use std::net::SocketAddr; use std::process::{Child, Command as ProcessCommand, Stdio}; -use runa_client::{LocalClient, WebSocketClient}; +use runa_client::WebSocketClient; use runa_protocol::{Command, Reply}; use tokio_tungstenite::tungstenite::{self, Message, client::IntoClientRequest}; fn start_server() -> (ServerProcess, SocketAddr) { - let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server")) - .args(["--listen", "127.0.0.1:0"]) + start_server_with(&["--listen", "127.0.0.1:0"], "invalid") +} + +fn start_server_with(args: &[&str], listen_env: &str) -> (ServerProcess, SocketAddr) { + let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .args(args) + .env("RUNA_LISTEN", listen_env) + .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() @@ -19,7 +25,7 @@ fn start_server() -> (ServerProcess, SocketAddr) { let mut startup = String::new(); stderr.read_line(&mut startup).unwrap(); let address = startup - .strip_prefix("runa-server: listening on ") + .strip_prefix("runa: listening on ") .expect("server should report its bound address") .trim() .parse() @@ -30,6 +36,43 @@ fn start_server() -> (ServerProcess, SocketAddr) { (server, address) } +#[test] +fn server_clap_supports_env_overrides_help_and_explicit_errors() { + let (_server, address) = start_server_with(&[], "127.0.0.1:0"); + let mut client = WebSocketClient::connect(address).unwrap(); + assert!( + client + .request(Command::GetBehaviors) + .unwrap() + .result + .is_ok() + ); + for args in [ + vec!["--listen", "invalid"], + vec!["--listen"], + vec!["--stdio"], + vec!["extra"], + vec!["--listen", "0.0.0.0:0"], + ] { + let output = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .args(args) + .env_remove("RUNA_LISTEN") + .output() + .unwrap(); + assert!(!output.status.success()); + assert!(!output.stderr.is_empty()); + } + let help = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) + .arg("--help") + .env_remove("RUNA_LISTEN") + .output() + .unwrap(); + assert!(help.status.success()); + let help = String::from_utf8(help.stdout).unwrap(); + assert!(help.contains("127.0.0.1:3737")); + assert!(help.contains("RUNA_LISTEN")); +} + fn watcher(address: SocketAddr) -> tungstenite::WebSocket { let mut request = format!("ws://{address}/").into_client_request().unwrap(); request @@ -195,7 +238,7 @@ fn websocket_rejects_binary_malformed_and_oversized_requests() { for message in [ Message::Binary(b"{}".to_vec().into()), Message::Text("{not-json}".into()), - Message::Text(" ".repeat(runa_server::MAX_REQUEST_LINE_BYTES + 1).into()), + Message::Text(" ".repeat(runa::MAX_MESSAGE_BYTES + 1).into()), ] { let mut socket = watcher(address); // An oversized frame may be rejected as soon as its header arrives, @@ -216,57 +259,9 @@ impl Drop for ServerProcess { } } -#[test] -fn local_client_launches_server_and_operates_the_same_live_image() { - let mut client = LocalClient::spawn(env!("CARGO_BIN_EXE_runa-server"), &[] as &[&str]).unwrap(); - - let published = client - .request(Command::PublishBehavior { - name: "counter".to_owned(), - expected_revision: None, - source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), - }) - .unwrap(); - assert_eq!(published.request_id, "runa-client-1"); - assert!(matches!(published.result, Ok(Reply::Behavior(_)))); - - let spawned = client - .request(Command::Spawn { - behavior_name: "counter".to_owned(), - state: Value::Integer(5), - mailbox_capacity: 2, - }) - .unwrap(); - assert_eq!(spawned.request_id, "runa-client-2"); - let actor_ref = match spawned.result { - Ok(Reply::ActorSpawned { actor_ref }) => actor_ref, - other => panic!("unexpected spawn response: {other:?}"), - }; - assert_eq!(actor_ref.len(), 21); - - client - .request(Command::Send { - actor_ref: actor_ref.clone(), - message: test_message(0, 7, None), - }) - .unwrap(); - client - .request(Command::Tick { - ticks: 100, - instruction_budget: 10, - }) - .unwrap(); - let inspected = client.request(Command::InspectActor { actor_ref }).unwrap(); - assert!(matches!( - inspected.result, - Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12) - )); - assert!(client.shutdown().unwrap().success()); -} - #[test] fn websocket_server_accepts_local_protocol_requests() { - let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server")) + let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa")) .args(["--listen", "127.0.0.1:0"]) .stdout(Stdio::null()) .stderr(Stdio::piped()) @@ -278,7 +273,7 @@ fn websocket_server_accepts_local_protocol_requests() { let mut startup = String::new(); stderr.read_line(&mut startup).unwrap(); let address = startup - .strip_prefix("runa-server: listening on ") + .strip_prefix("runa: listening on ") .expect("server should report its bound address") .trim() .parse::() @@ -408,14 +403,3 @@ fn websocket_server_accepts_local_protocol_requests() { actor_ref: echo, }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested && actor.reserved.is_none())); } - -fn test_message(kind: i64, value: i64, target: Option) -> Value { - Value::Record(std::collections::BTreeMap::from([ - ("kind".to_owned(), Value::Integer(kind)), - ("value".to_owned(), Value::Integer(value)), - ( - "target".to_owned(), - target.map_or(Value::Null, Value::ActorRef), - ), - ])) -} diff --git a/docs/architecture.md b/docs/architecture.md index 6e7aeed..080d360 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -63,7 +63,7 @@ Rust supplies a minimal, reliable substrate; increasingly much of the actor fram +-----------------------------------------------------------+ ``` -The graphical world should use internal service calls to the image API. The VS Code extension connects to the same live image through a versioned network transport. CLI tools, tests, and local automation can use in-process calls or a child-process protocol over stdio. The initial WebSocket transport is unauthenticated and must bind only to loopback; a remote-capable network boundary must authenticate and authorize clients. Neither transport creates a separate actor runtime or image. +The graphical world should use internal service calls to the image API. The VS Code extension, CLI tools, tests, and local automation connect to the same live image through the versioned runa-debug WebSocket transport; trusted embedded callers can use in-process calls. The initial WebSocket transport is unauthenticated and must bind only to loopback; a remote-capable network boundary must authenticate and authorize clients. Client connections do not create a separate actor runtime or image. ## 4. Fundamental Actor Model @@ -201,7 +201,7 @@ The graphical inspector should support selecting a component, locating its ownin ## 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. The extension uses a network transport to reach the same running image; local stdio remains available for child-process integrations. Neither transport changes 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 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. Suggested operation families (provisional names): @@ -235,11 +235,11 @@ These are desired capabilities, not a requirement to implement every endpoint im ### 8.1 Security -Normal actors should not automatically receive development privileges. Arbitrary evaluation, state mutation, code replacement, and process control require explicit capabilities. The stdio bridge is intended for a trusted parent process. The initial WebSocket bridge accepts unauthenticated connections only on IPv4 or IPv6 loopback and grants each connection the full development API; any local process may connect. It is suitable only for trusted local development. Never bind this endpoint to a non-loopback address or expose it remotely until authentication and authorization are implemented. +Normal actors should not automatically receive development privileges. Arbitrary evaluation, state mutation, code replacement, and process control require explicit capabilities. The initial WebSocket bridge accepts unauthenticated connections only on IPv4 or IPv6 loopback and grants each connection the full development API; any local process may connect. It is suitable only for trusted local development. Never bind this endpoint to a non-loopback address or expose it remotely until authentication and authorization are implemented. ## 9. VS Code Extension -VS Code is an external client of the same live image, not a separate runtime. The initial development transport is versioned JSON over runa-debug WebSockets, bound to loopback and unauthenticated; the extension must connect only to a trusted local image. Remote connections remain out of scope until authentication and authorization are implemented. The extension does not host actors or create a separate image. The local stdio protocol remains useful for CLI tools, tests, and other child-process clients. +VS Code is an external client of the same live image, not a separate runtime. The initial development transport is versioned JSON over runa-debug WebSockets, bound to loopback and unauthenticated; the extension must connect only to a trusted local image. Remote connections remain out of scope until authentication and authorization are implemented. The extension does not host actors or create a separate image. The REPL uses the same WebSocket protocol. The first extension increment implements explicit connect/disconnect, an actor explorer, and read-only state/mailbox/execution snapshots. Server `image_changed` @@ -291,12 +291,9 @@ The Rust workspace uses **acyclic, inward-pointing dependencies**. The executabl Arrows mean **"depends on"**, not runtime data flow: ```text - runa - / | \ - v v v - runa-world runa-server - | / \ - v v v + runa-world runa runa-cli + | / \ | + v v v v runa-image-api runa-protocol | | v | @@ -310,8 +307,9 @@ Arrows mean **"depends on"**, not runtime data flow: runa-world also depends on runa-graphics (platform drawing/input contracts); runa-graphics depends only on runa-core when needed. -runa-server depends on runa-image-api and runa-protocol. -runa wires together world, server, runtime, storage and platform. +runa depends on runa-image-api and runa-protocol. +runa-cli uses runa-client and runa-protocol to connect to the image. +World, storage and platform integration remain planned. ``` The diagram is conceptual; Cargo dependencies should be kept minimal. A dependency is permitted only when the importing crate actually uses that crate's public contracts. The VS Code extension is a separate TypeScript project and depends on the **wire protocol**, not Rust crates. @@ -327,11 +325,11 @@ The diagram is conceptual; Cargo dependencies should be kept minimal. A dependen | `runa-runtime` | Scheduling, actor execution coordination, state commits, crash recovery, binding actor semantics to Rune execution | `runa-core`, `runa-actor`, `runa-code`, `runa-rune` | | `runa-image-api` | Trusted in-process application services for spawning, inspecting, sending, editing, and upgrading | `runa-core`, `runa-actor`, `runa-code`, `runa-runtime` | | `runa-protocol` | Versioned transport-neutral request/response/event DTOs and validation | `serde` | -| `runa-server` | Protocol dispatch, JSON-lines stdio, and loopback-only WebSocket transport | `runa-protocol`, `runa-image-api` | -| `runa-client` | Local child-process stdio and loopback runa-debug WebSocket clients | `runa-protocol` | +| `runa` | Image startup, configuration, protocol dispatch, and loopback-only WebSocket transport | `runa-protocol`, `runa-image-api` | +| `runa-cli` | Interactive REPL and connection argument parsing | `runa-client`, `runa-protocol` | +| `runa-client` | Loopback runa-debug WebSocket client | `runa-protocol` | | `runa-graphics` | Native window, rendering and input mechanisms; no actor policy | `runa-core` if necessary | | `runa-world` | Morphic-style graphical composition and in-world development UI | `runa-image-api`, `runa-graphics` | -| `runa` | Startup, configuration, resource wiring, shutdown | Top-level crates it instantiates | The current prototype has added `runa-code` for exact-source identity and in-memory compilation caching, and `runa-image-api` as the trusted, @@ -345,15 +343,15 @@ publication, history and rollback, actor lifecycle and messaging, snapshots, ticks, and event retrieval. The CLI uses these application services rather than reaching into the code store and runtime directly. `runa-protocol` now defines versioned, serializable request, response, and event DTOs without -depending on a transport or the runtime implementation. `runa-server` now +depending on a transport or the runtime implementation. `runa` now dispatches those commands through the image API and maps service results into -protocol replies and typed errors. The dispatcher is exposed through local -child-process JSON-lines stdio and an unauthenticated WebSocket listener restricted -to IPv4 or IPv6 loopback; both enforce a 1 MiB request limit. The WebSocket +protocol replies and typed errors. The dispatcher is exposed through an +unauthenticated WebSocket listener restricted +to IPv4 or IPv6 loopback, enforcing a 1 MiB request limit. The WebSocket listener shares one image across at most 16 concurrent clients. Every local client receives full development API access, so the endpoint is for trusted local development only. In-process consumers call `runa-image-api`, and -external local tools can use `runa-client` to launch the stdio server. Actors +external local tools can use `runa-client` to connect to the server. Actors do not depend on network transport to coexist in one live image. Remote network access will require authentication and authorization. @@ -393,7 +391,7 @@ runa/ │ ├── runa-runtime/ │ ├── runa-image-api/ │ ├── runa-protocol/ -│ ├── runa-server/ +│ ├── runa-cli/ │ ├── runa-client/ │ ├── runa-graphics/ │ ├── runa-world/ @@ -465,16 +463,20 @@ mapping, and local-variable inspection remain future work. The first Phase 5 increments are a trusted in-process application API used by the CLI, a transport-neutral versioned protocol contract, and a dispatcher -exposed through bounded local JSON-lines stdio. The interactive CLI now uses +exposed through bounded loopback WebSockets. The interactive `runa-cli` uses `runa-client` to connect over loopback WebSockets to a server-owned image, rather than embedding a runtime. Multiple clients share actors and behavior revisions; -client disconnect/reconnect leaves the running image intact. `runa-client` -also launches the stdio server for external local tools. The server provides an +client disconnect/reconnect leaves the running image intact. The `runa` server provides an unauthenticated runa-debug WebSocket endpoint restricted to loopback for local VS Code development. Remote access will require authentication and authorization; neither network transport nor client location changes the single-image actor model. +Both binaries use Clap derive/env argument parsing. `runa` starts the server +at `127.0.0.1:3737` by default (`--listen` / `RUNA_LISTEN`), and `runa-cli` +connects to that address (`--connect` / `RUNA_CONNECT`). Flags override environment +variables. Mise's `start` and `cli` tasks run the server and REPL respectively. + ### Phase 6 — Graphical world - Native window/rendering, input, scene manager, Rune-driven button and window. diff --git a/docs/protocol.md b/docs/protocol.md index 46d3984..212f378 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -1,10 +1,12 @@ # Runa Protocol `runa-protocol` defines versioned request, response, and event data contracts -for clients and the `runa-server` dispatcher. It is independent of transport -and runtime implementation. `runa-server` accepts JSON Lines over stdio or, -when started with `--listen`, as WebSocket text messages bound to an IPv4 or -IPv6 loopback address. The listener is unauthenticated and intended only for +for clients and the `runa` dispatcher. It is independent of transport +and runtime implementation. `runa` accepts WebSocket text messages bound to an IPv4 or +IPv6 loopback address, defaulting to `127.0.0.1:3737` without arguments. +Clap parses `--listen` / `RUNA_LISTEN` on the server and `--connect` / +`RUNA_CONNECT` on `runa-cli`; explicit flags override environment variables. +The listener is unauthenticated and intended only for trusted local development. Both the REPL and VS Code use WebSockets. ## Version 3 @@ -84,17 +86,17 @@ any of its sends. `get_events` drains and returns the events currently accumulated by this single-worker image. It is not a durable subscription or replayable event log. -Both transports limit each request to 1 MiB before parsing. Malformed +The transport limits each request to 1 MiB before parsing. Malformed JSON or an oversized message terminates that connection with an error; no protocol -response is emitted when a request cannot be decoded. Stdio EOF terminates the -process, while a WebSocket disconnect closes only that client connection. The listener +response is emitted when a request cannot be decoded. A WebSocket disconnect +closes only that client connection. The listener allows up to 16 simultaneous clients and serializes each dispatch through the same image. WebSocket response payloads are capped at 1 MiB, with one in-flight dispatch per client and a bounded shared dispatch queue. Every local client can use the full development API; loopback binding is not authentication or authorization. Remote access must not be enabled until appropriate authentication and authorization exist. -The `runa-client` crate launches the stdio server locally and validates -response protocol versions and request IDs. Its `WebSocketClient` connects to +The `runa-client` crate validates response protocol versions and request IDs. +Its `WebSocketClient` connects to loopback endpoints using bounded WebSocket messages and correlation checks, validating interleaved notifications while waiting for responses. Connection establishment has a 5-second timeout; WebSocket writes and response @@ -106,7 +108,7 @@ The interactive REPL uses WebSockets and never owns the image. Disconnecting cli does not stop actors or reset behavior revisions; reconnecting clients can retrieve them through `list_actors` and `get_behaviors`. The network executable publishes the bundled `counter` behavior on image startup, not on connection. -The stdio executable continues to start with an empty image. +The stdio transport and child-process client have been removed. The TypeScript VS Code client currently issues only `list_actors` and `inspect_actor`. It enforces loopback endpoints, bounded UTF-8 text messages, @@ -123,7 +125,7 @@ the actor tree and open snapshot documents. Manual refresh remains available. Connect to `ws://127.0.0.1:3737/` (or a configured numeric loopback endpoint) and offer the `runa-debug` WebSocket subprotocol. The server requires and echoes that subprotocol. Plain TCP JSON Lines is no longer accepted by the network -listener. The request/response contract remains version 3; stdio is unchanged. +listener. The request/response contract remains version 3. Each WebSocket text message contains exactly one JSON request, response, or notification, without newline framing. Binary messages and malformed requests terminate the connection. Fragmented text is supported with a 1 MiB aggregate diff --git a/docs/runtime.md b/docs/runtime.md index a0b746d..2799c4d 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -3,11 +3,15 @@ Start the server-owned, in-memory image: ```sh -mise run serve-network +mise run start ``` -Then connect from another terminal with `cargo run -p runa`, or use -`cargo run -p runa -- --connect ` for a custom listener. +Then connect from another terminal with `mise run cli`, or use +`cargo run -p runa-cli -- --connect ` for a custom listener. +The server (`runa`) defaults to `127.0.0.1:3737`; `--listen` or `RUNA_LISTEN` +select another loopback address. The REPL (`runa-cli`) accepts `--connect` or +`RUNA_CONNECT`, defaulting to the same address. Explicit flags override environment +variables. Both executables provide Clap-generated `--help` and `--version`. Quitting the REPL leaves the image running. Reconnecting clients can list and inspect the same actors; local bindings and history do not survive a REPL restart. Server exit still loses the entire image. @@ -26,7 +30,7 @@ The CLI sends versioned requests to the loopback-only WebSocket listener using the `runa-debug` subprotocol. VS Code uses the same protocol and automatically refreshes on non-destructive server change notifications. These notifications do not tick the image or drain the REPL's explicit `events` queue. -The separate stdio endpoint starts an empty image for child-process clients. +There is no stdio transport; all external clients use WebSockets. Example: @@ -79,12 +83,12 @@ also offers references inside a `"$actor"` string. trusted in-process service boundary. - `runa-protocol` defines versioned transport-neutral request, response, and event DTOs; it does not dispatch requests or access runtime state. -- `runa-server` dispatches those requests through `runa-image-api` and maps - results to protocol replies; its executable accepts newline-delimited JSON - over stdin/stdout or JSON text messages over loopback WebSockets, with a +- `runa` dispatches those requests through `runa-image-api` and maps + results to protocol replies; its executable accepts JSON text messages over + loopback WebSockets, with a 1 MiB request limit. -- `runa-client` provides correlated, bounded stdio and loopback WebSocket clients. -- `runa` provides an interactive CLI using the WebSocket client, without an embedded +- `runa-client` provides a correlated, bounded loopback WebSocket client. +- `runa-cli` provides an interactive CLI using the WebSocket client, without an embedded runtime. Connection errors are explicit; requests are never auto-retried. The image API is process-local and intended for trusted callers. It has no diff --git a/mise.toml b/mise.toml index a03904d..32b23e8 100644 --- a/mise.toml +++ b/mise.toml @@ -18,16 +18,12 @@ description = "Test the extension in VS Code (requires a display or xvfb-run)" run = "npm --prefix vscode run test:host" [tasks.start] -description = "Start the local Runa CLI" +description = "Start the Runa image server on localhost" run = "cargo run -p runa" -[tasks.serve] -description = "Run the local JSON-lines protocol server" -run = "cargo run -p runa-server" - -[tasks.serve-network] -description = "Run the runa-debug WebSocket server on localhost" -run = "cargo run -p runa-server -- --listen 127.0.0.1:3737" +[tasks.cli] +description = "Connect the Runa CLI to the image server" +run = "cargo run -p runa-cli" [tasks.lint] description = "Check Rust targets and the TypeScript extension" diff --git a/vscode/README.md b/vscode/README.md index 481f2f4..0a7b755 100644 --- a/vscode/README.md +++ b/vscode/README.md @@ -13,7 +13,7 @@ From the repository root: ```sh npm --prefix vscode ci -mise run serve-network +mise run start ``` Open the repository in VS Code. In Run and Debug, select **Run Runa Extension** @@ -24,7 +24,7 @@ In that host: 1. Run **Runa: Connect to Local Image**. The default is `127.0.0.1:3737`; `runa.address` changes the suggested endpoint. -2. Expand **Runa Actors** in Explorer. Use the REPL (`mise run start`) to +2. Expand **Runa Actors** in Explorer. Use the REPL (`mise run cli`) to spawn actors in the same server. The tree updates automatically. 3. Click an actor or run **Runa: Inspect Actor** to open its JSON snapshot. 4. Run **Runa: Disconnect** to close only this client connection. diff --git a/vscode/integration/repl.test.ts b/vscode/integration/repl.test.ts index f02b382..6e4da28 100644 --- a/vscode/integration/repl.test.ts +++ b/vscode/integration/repl.test.ts @@ -32,9 +32,10 @@ test( const executable = resolve( __dirname, "../../../target/debug", - process.platform === "win32" ? "runa.exe" : "runa", + process.platform === "win32" ? "runa-cli.exe" : "runa-cli", ); - const repl = spawn(executable, ["--connect", image.address], { + const repl = spawn(executable, [], { + env: { ...process.env, RUNA_CONNECT: image.address }, stdio: ["pipe", "pipe", "pipe"], }); t.after(() => { @@ -69,5 +70,16 @@ test( assert.deepEqual(actor.state, { type: "integer", data: 12n }); assert.deepEqual(actor.queued, []); assert.equal(actor.reserved, null); + const overridden = spawn(executable, ["--connect", image.address], { + env: { ...process.env, RUNA_CONNECT: "invalid" }, + stdio: ["pipe", "pipe", "pipe"], + }); + t.after(() => { + if (overridden.exitCode === null && overridden.signalCode === null) overridden.kill(); + }); + const overriddenExit = once(overridden, "exit"); + overridden.stdin.end("actors\nquit\n"); + const [overrideCode] = await overriddenExit; + assert.equal(overrideCode, 0); }, ); diff --git a/vscode/integration/server.ts b/vscode/integration/server.ts index 3904420..da9b1ee 100644 --- a/vscode/integration/server.ts +++ b/vscode/integration/server.ts @@ -10,7 +10,7 @@ export async function startImage(): Promise<{ address: string; stop: () => Promi const executable = resolve( __dirname, "../../../target/debug", - process.platform === "win32" ? "runa-server.exe" : "runa-server", + process.platform === "win32" ? "runa.exe" : "runa", ); const child = spawn(executable, ["--listen", "127.0.0.1:0"], { stdio: ["ignore", "ignore", "pipe"], @@ -41,7 +41,7 @@ export async function startImage(): Promise<{ address: string; stop: () => Promi child.once("exit", exited); lines.on("line", (line) => { diagnostics = (diagnostics + line + "\n").slice(-8192); - const address = line.match(/^runa-server: listening on (.+)$/)?.[1]; + const address = line.match(/^runa: listening on (.+)$/)?.[1]; if (address) { cleanup(); resolve(address); diff --git a/vscode/package.json b/vscode/package.json index 15c030e..3cb3a62 100644 --- a/vscode/package.json +++ b/vscode/package.json @@ -20,8 +20,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-server -p runa && npm run compile && node --test out/integration/*.test.js", - "test:host": "cargo build --manifest-path ../Cargo.toml -p runa-server && 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", diff --git a/vscode/src/extension.ts b/vscode/src/extension.ts index 8b04bc8..c2801cb 100644 --- a/vscode/src/extension.ts +++ b/vscode/src/extension.ts @@ -176,8 +176,7 @@ export function activate(context: vscode.ExtensionContext): void { (await vscode.window.showInputBox({ title: "Connect to a trusted local Runa image (no authentication)", value: vscode.workspace.getConfiguration("runa").get("address", "127.0.0.1:3737"), - prompt: - "Start the server with mise run serve-network. The extension does not start an image.", + prompt: "Start the server with mise run start. The extension does not start an image.", validateInput: (value) => { try { parseEndpoint(value);