diff --git a/CHANGELOG.md b/CHANGELOG.md index d9e5230..c91f561 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ All notable changes to this project will be documented in this file. ### Changed +- **Breaking:** Protocol v5 adds actor log thresholds to snapshots, guarded + editing operations, and opt-in log-history/follow operations and notifications. - Move image sources and actors into a dedicated Runa Activity Bar sidebar, with the source file tree first and both views visible by default. - Remove the VS Code Add Source Folder to Workspace command; image files are @@ -30,6 +32,11 @@ All notable changes to this project will be documented in this file. ### Added +- Guarded actor state and reserved/queued message editing through the CLI, + REPL, and VS Code for paused/stopped/failed actors without active execution. +- Immediate actor logs via `ctx.log(level, message)` and severity helpers, + per-actor thresholds, bounded retained history, and server-filtered CLI + `logs [-f|--follow] [actor...]` subscriptions. - Mutable image-owned source files organized beneath `source/`, with independent commit history and generated readable revision IDs. Spawn compiles the current working file whether committed or not. diff --git a/Cargo.lock b/Cargo.lock index 9123dfb..38201d7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -119,6 +119,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block2" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5" +dependencies = [ + "objc2", +] + [[package]] name = "bytes" version = "1.12.1" @@ -252,6 +261,17 @@ dependencies = [ "typenum", ] +[[package]] +name = "ctrlc" +version = "3.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0b1fab2ae45819af2d0731d60f2afe17227ebb1a1538a236da84c93e9a60162" +dependencies = [ + "dispatch2", + "nix 0.31.3", + "windows-sys 0.61.2", +] + [[package]] name = "data-encoding" version = "2.11.1" @@ -268,6 +288,18 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "dispatch2" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" +dependencies = [ + "bitflags", + "block2", + "libc", + "objc2", +] + [[package]] name = "endian-type" version = "0.1.2" @@ -566,6 +598,18 @@ dependencies = [ "libc", ] +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -647,6 +691,21 @@ dependencies = [ "autocfg", ] +[[package]] +name = "objc2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08849bbd4767dfae9457696856ae1c84fe4e0281bbe4a7abff2d0e06fb7981f8" +dependencies = [ + "objc2-encode", +] + +[[package]] +name = "objc2-encode" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" + [[package]] name = "once_cell" version = "1.21.4" @@ -842,6 +901,7 @@ name = "runa-cli" version = "0.0.0" dependencies = [ "clap", + "ctrlc", "runa", "runa-client", "runa-core", @@ -1043,7 +1103,7 @@ dependencies = [ "libc", "log", "memchr", - "nix", + "nix 0.29.0", "radix_trie", "unicode-segmentation", "unicode-width 0.2.3", diff --git a/README.md b/README.md index de2c542..4d10e28 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,7 @@ lists, records, and distinct actor-reference values. `spawn ` sets initial state; `send ` sends any supported value. JSON objects of the form `{"$actor": "name-or-@reference"}` embed an actor reference, including inside a record or list. See the [runtime guide](docs/runtime.md) -for Rune conversions and value limits. Protocol version 4 uses +for Rune conversions and value limits. Protocol version 5 uses `handle(ctx, state, sender, message) -> new_state`; `ctx.send(recipient, message)` records sends with the calling actor as sender, and `ctx.send_from(sender, recipient, message)` supports forwarding. @@ -189,6 +189,39 @@ leaves actors in the server. Detailed runtime event subscriptions and structured compilation diagnostics are not implemented yet. See the [extension guide](vscode/README.md) for connection restrictions, snapshot semantics, and tests. +## Repairing actors and reading logs + +State and messages can be edited only on paused, stopped, or failed actors with +no active invocation. A paused suspended handler must be killed first. State +edits preserve messages; message edits preserve state. Neither restarts an actor. +Reserved messages are retried first on restart; clearing the reserved array +removes the failed input. + +```text +set-state @ 1 +set-messages @ {"reserved":[],"queued":[{"op":"add","value":2}]} +log-level @ debug +logs @ +logs -f +``` + +These are REPL commands and `runa-cli` subcommands (quote JSON when using a shell). +REPL log filters also accept session-local actor names. Omitting the filter selects all +actors. `logs` prints retained history; `logs -f` additionally subscribes to +future entries until Ctrl+C, which returns to the REPL or closes a one-shot CLI. +In VS Code, actor context menus offer **Edit Actor State...**, +**Edit Reserved and Queued Messages...**, and **Set Actor Log Level...**. + +Rune handlers can call `ctx.log("info", "message")`, or `ctx.trace`, `ctx.debug`, +`ctx.info`, `ctx.warn`, `ctx.error`, and `ctx.fatal` with a string message. +Severity order is trace < debug < info < warn < error < fatal; the default +threshold is info. Only entries at or above the actor's threshold are retained +or emitted. Logs are immediate diagnostics, not transactional effects: they +survive handler failure and may repeat on retries. Each actor keeps its latest +1,000 accepted messages in memory, each limited to 512 UTF-8 bytes. Restart +preserves logs and the threshold. Server exit loses them. Rebuild the server, +CLI, and extension together for protocol v5. + ## Development Install [mise](https://mise.jdx.dev/) and run `mise install` to install Tombi. diff --git a/crates/runa-actor/src/lib.rs b/crates/runa-actor/src/lib.rs index e54ece7..0a7117d 100644 --- a/crates/runa-actor/src/lib.rs +++ b/crates/runa-actor/src/lib.rs @@ -38,6 +38,7 @@ pub enum Status { #[derive(Debug, Clone)] pub struct ActorSnapshot { + pub log_level: runa_core::LogLevel, pub actor_ref: ActorRef, pub mailbox: MailboxId, pub implementation: ImplementationId, diff --git a/crates/runa-cli/Cargo.toml b/crates/runa-cli/Cargo.toml index 3df9206..54da80b 100644 --- a/crates/runa-cli/Cargo.toml +++ b/crates/runa-cli/Cargo.toml @@ -5,6 +5,7 @@ edition = "2024" [dependencies] clap.workspace = true +ctrlc = "3.4" runa-client = { path = "../runa-client" } runa-core = { path = "../runa-core" } runa-protocol = { path = "../runa-protocol" } diff --git a/crates/runa-cli/src/main.rs b/crates/runa-cli/src/main.rs index 654a771..8ba3ea8 100644 --- a/crates/runa-cli/src/main.rs +++ b/crates/runa-cli/src/main.rs @@ -8,7 +8,7 @@ use std::num::NonZeroUsize; #[cfg(test)] use runa::Server; use runa_client::WebSocketClient; -use runa_core::ActorRef; +use runa_core::{ActorRef, LogLevel}; use runa_protocol::{Command, Reply, SystemMessageDto, Value}; use rustyline::completion::{Completer, Pair}; use rustyline::error::ReadlineError; @@ -20,14 +20,21 @@ use rustyline::{Context, Editor, Helper}; const MAILBOX_CAPACITY: usize = 64; const DEFAULT_TICK_BUDGET: usize = 32; +static INTERRUPTED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); #[cfg(test)] const COUNTER_BEHAVIOR: &str = "actors/counter"; trait Connection { fn dispatch(&mut self, command: Command) -> Result; + fn next_log(&mut self) -> Result, String> { + Err("log following requires a WebSocket connection".to_owned()) + } } impl Connection for WebSocketClient { + fn next_log(&mut self) -> Result, String> { + WebSocketClient::next_log(self).map_err(|error| error.to_string()) + } fn dispatch(&mut self, command: Command) -> Result { self.request(command) .map_err(|error| error.to_string())? @@ -74,9 +81,32 @@ impl ReplHelper { fn candidates(&self, line: &str, position: usize) -> Vec { const COMMANDS: &[&str] = &[ - "ps", "sources", "events", "exit", "help", "history", "inspect", "become", "kill", - "reload", "write", "let", "pause", "commit", "quit", "restart", "resume", "rollback", - "send", "spawn", "stop", "tick", + "set-state", + "set-messages", + "log-level", + "logs", + "ps", + "sources", + "events", + "exit", + "help", + "history", + "inspect", + "become", + "kill", + "reload", + "write", + "let", + "pause", + "commit", + "quit", + "restart", + "resume", + "rollback", + "send", + "spawn", + "stop", + "tick", ]; let (start, reference_value) = Self::completion_start(line, position); if reference_value { @@ -104,6 +134,12 @@ impl ReplHelper { | ["kill"] | ["reload"] | ["become"] => self.actor_candidates(), + ["set-state"] | ["set-messages"] | ["log-level"] => self.actor_candidates(), + ["log-level", _] => ["trace", "debug", "info", "warn", "error", "fatal"] + .into_iter() + .map(str::to_owned) + .collect(), + ["logs", ..] => self.actor_candidates(), ["become", _] | ["history"] | ["rollback"] | ["commit"] | ["write"] => { self.behaviors.clone() } @@ -188,6 +224,26 @@ struct Args { #[derive(Debug, clap::Subcommand)] enum CliCommand { + /// Replace committed state on a paused, stopped, or failed actor without active execution. + SetState { + actor: String, + #[arg(allow_hyphen_values = true)] + state: String, + }, + /// Replace reserved and queued messages using {"reserved":[],"queued":[]} JSON. + SetMessages { actor: String, messages: String }, + /// Set the actor's minimum accepted log severity. + LogLevel { + actor: String, + #[arg(value_parser = LogLevel::parse)] + level: LogLevel, + }, + /// Show retained logs, optionally following future entries (Ctrl+C to stop). + Logs { + #[arg(short, long)] + follow: bool, + actors: Vec, + }, /// Start the interactive REPL. Repl, /// List actors in the image. @@ -257,6 +313,21 @@ impl CliCommand { }) }; Ok(match self { + Self::SetState { actor, state: json } => { + edit_request(actor_ref(&actor)?, Some(parse_value(&json, &state)?), None)? + } + Self::SetMessages { actor, messages } => edit_request( + actor_ref(&actor)?, + None, + Some(parse_value(&messages, &state)?), + )?, + Self::LogLevel { actor, level } => Command::SetLogLevel { + actor_ref: actor_ref(&actor)?, + level, + }, + Self::Logs { .. } => { + return Err("logs requires the streaming command handler".to_owned()); + } Self::Ps => Command::ListActors, Self::Sources => Command::ListSources, Self::Spawn { path, state: json } => Command::Spawn { @@ -299,6 +370,9 @@ fn run_command( server: &mut dyn Connection, output: &mut impl Write, ) -> Result<(), Box> { + if let CliCommand::Logs { follow, actors } = command { + return show_logs(server, output, actors, follow).map_err(Into::into); + } let request = command.request(server)?; let scheduled = matches!( &request, @@ -349,6 +423,7 @@ fn run_command( fn main() -> Result<(), Box> { let args = ::parse(); + ctrlc::set_handler(|| INTERRUPTED.store(true, std::sync::atomic::Ordering::SeqCst))?; if matches!(args.command, CliCommand::Quit) { return Ok(()); } @@ -467,6 +542,64 @@ fn execute_line( let command = words[0]; let args = &words[1..]; let result = match command { + "set-state" | "set-messages" => { + let result = (|| -> Result<_, String> { + let actor = args.first().ok_or("expected an actor reference and JSON")?; + let actor = parse_actor_argument(&[actor], state)?; + let value = parse_value(after_words(line, 2), state)?; + let request = if command == "set-state" { + edit_request(actor, Some(value), None)? + } else { + edit_request(actor, None, Some(value))? + }; + dispatch(server, request) + })(); + match result { + Ok(Reply::Acknowledged) => writeln!(output, "ok"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + "log-level" => { + let result = (|| -> Result<_, String> { + let [actor, level] = args else { + return Err("usage: log-level ".to_owned()); + }; + dispatch( + server, + Command::SetLogLevel { + actor_ref: parse_actor_argument(&[actor], state)?, + level: LogLevel::parse(level)?, + }, + ) + })(); + match result { + Ok(Reply::Acknowledged) => writeln!(output, "ok"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + "logs" => { + let parsed = ::try_parse_from( + std::iter::once("runa-cli").chain(words.iter().copied()), + ); + let result = (|| -> Result<(), String> { + let CliCommand::Logs { actors, follow } = + parsed.map_err(|error| error.to_string())?.command + else { + unreachable!() + }; + let actors = actors + .iter() + .map(|actor| parse_actor_argument(&[actor], state)) + .collect::, _>>()?; + show_logs(server, output, actors, follow) + })(); + match result { + Ok(()) => Ok(()), + Err(error) => writeln!(output, "error: {error}"), + } + } "help" => print_help(output), "spawn" => spawn_command(output, line, server, state), "sources" if args.is_empty() => match dispatch(server, Command::ListSources) { @@ -820,6 +953,109 @@ fn dispatch(server: &mut dyn Connection, command: Command) -> Result, + messages: Option, +) -> Result { + let messages = if let Some(messages) = messages { + let Value::Record(mut fields) = messages else { + return Err("messages must be an object with reserved and queued arrays".to_owned()); + }; + let (Some(Value::List(mut reserved)), Some(Value::List(queued))) = + (fields.remove("reserved"), fields.remove("queued")) + else { + return Err("messages must contain reserved and queued arrays".to_owned()); + }; + if !fields.is_empty() || reserved.len() > 1 { + return Err( + "reserved must have zero or one message; unknown fields are forbidden".to_owned(), + ); + } + Some(runa_protocol::MailboxEdit { + reserved: reserved.pop(), + queued, + }) + } else { + None + }; + Ok(Command::EditActor { + actor_ref, + state, + messages, + }) +} + +fn show_logs( + server: &mut dyn Connection, + output: &mut impl Write, + actors: Vec, + follow: bool, +) -> Result<(), String> { + let actors = actors + .iter() + .map(|actor| parse_actor_argument(&[actor], &ReplState::default())) + .collect::, _>>()?; + let mut after = 0; + loop { + let Reply::Logs { + entries, + next_sequence, + has_more, + } = dispatch( + server, + Command::Logs { + actors: actors.clone(), + follow, + after, + }, + )? + else { + return Err("expected log history".to_owned()); + }; + for entry in entries { + print_log(output, &entry)?; + } + after = next_sequence; + if !has_more { + break; + } + } + INTERRUPTED.store(false, std::sync::atomic::Ordering::SeqCst); + let result = (|| { + while follow && !INTERRUPTED.load(std::sync::atomic::Ordering::SeqCst) { + if let Some(entry) = server.next_log()? + && entry.sequence > after + { + after = entry.sequence; + print_log(output, &entry)?; + } + } + Ok(()) + })(); + if follow { + dispatch( + server, + Command::Logs { + actors, + follow: false, + after, + }, + )?; + } + result +} + +fn print_log(output: &mut impl Write, entry: &runa_core::LogEntry) -> Result<(), String> { + writeln!( + output, + "{} @{} {:?}: {}", + entry.sequence, entry.actor_ref, entry.level, entry.message + ) + .and_then(|()| output.flush()) + .map_err(|error| error.to_string()) +} + fn unexpected_reply(output: &mut W, reply: Reply) -> io::Result<()> { writeln!(output, "error: unexpected protocol response: {reply:?}") } @@ -934,6 +1170,10 @@ fn print_help(output: &mut W) -> io::Result<()> { "Commands:\n\ let = spawn \n\ spawn \n\ + set-state \n\ + set-messages \n\ + log-level \n\ + logs [-f|--follow] [actor...]\n\ sources\n\ write \n\ commit \n\ diff --git a/crates/runa-cli/tests/arguments.rs b/crates/runa-cli/tests/arguments.rs index e3676d6..77ba287 100644 --- a/crates/runa-cli/tests/arguments.rs +++ b/crates/runa-cli/tests/arguments.rs @@ -14,9 +14,30 @@ fn cli_help_and_version_are_available_without_connecting() { assert!(help.contains("RUNA_CONNECT")); assert!(help.contains("127.0.0.1:3737")); for subcommand in [ - "repl", "ps", "sources", "spawn", "send", "tick", "inspect", "events", "write", "commit", - "history", "rollback", "stop", "kill", "restart", "pause", "resume", "reload", "become", + "repl", + "ps", + "sources", + "spawn", + "send", + "tick", + "inspect", + "events", + "write", + "commit", + "history", + "rollback", + "stop", + "kill", + "restart", + "pause", + "resume", + "reload", + "become", "quit", + "set-state", + "set-messages", + "log-level", + "logs", ] { let output = Command::new(env!("CARGO_BIN_EXE_runa-cli")) .args([subcommand, "--help"]) diff --git a/crates/runa-client/src/lib.rs b/crates/runa-client/src/lib.rs index 341da5f..ac57a92 100644 --- a/crates/runa-client/src/lib.rs +++ b/crates/runa-client/src/lib.rs @@ -19,6 +19,7 @@ pub struct WebSocketClient { next_request_id: u64, revision: u64, disconnected: bool, + log_queue: std::collections::VecDeque, } impl WebSocketClient { @@ -69,6 +70,7 @@ impl WebSocketClient { next_request_id: 0, revision: 0, disconnected: false, + log_queue: std::collections::VecDeque::new(), }) } @@ -80,7 +82,14 @@ impl WebSocketClient { ) .into()); } + let changes_subscription = matches!(command, ProtocolCommand::Logs { .. }); let result = self.request_inner(command); + if changes_subscription + && matches!(&result, Ok(response) if matches!(&response.result, Ok(runa_protocol::Reply::Logs { .. }))) + { + // Events preceding this reply belong to the previous subscription. + self.log_queue.clear(); + } if result.is_err() { self.disconnected = true; // A lost response may follow a committed operation. Never retry it @@ -103,6 +112,49 @@ impl WebSocketClient { result } + pub fn next_log(&mut self) -> Result, ClientError> { + if let Some(entry) = self.log_queue.pop_front() { + return Ok(Some(entry)); + } + self.socket + .get_ref() + .set_read_timeout(Some(Duration::from_millis(100)))?; + loop { + match self.socket.read() { + Ok(Message::Text(text)) => { + let event: Notification = serde_json::from_str(&text)?; + if event.version != PROTOCOL_VERSION { + return Err(ClientError::UnsupportedVersion(event.version)); + } + match event.event { + ImageEvent::Log(entry) => return Ok(Some(entry)), + ImageEvent::ImageChanged { revision } => self.revision = revision, + } + } + Ok(Message::Ping(_)) => self.socket.flush().map_err(io::Error::other)?, + Ok(Message::Pong(_)) => {} + Err(tungstenite::Error::Io(error)) + if matches!( + error.kind(), + io::ErrorKind::TimedOut + | io::ErrorKind::WouldBlock + | io::ErrorKind::Interrupted + ) => + { + return Ok(None); + } + Err(error) => return Err(io::Error::other(error).into()), + _ => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "log connection closed", + ) + .into()); + } + } + } + } + fn close_inner(&mut self) -> io::Result<()> { let timeout = Duration::from_secs(5); let deadline = Instant::now() + timeout; @@ -159,7 +211,17 @@ impl WebSocketClient { if event.version != PROTOCOL_VERSION { return Err(ClientError::UnsupportedVersion(event.version)); } - let ImageEvent::ImageChanged { revision } = event.event; + let ImageEvent::ImageChanged { revision } = event.event else { + if let ImageEvent::Log(entry) = event.event { + if self.log_queue.len() >= 1024 { + return Err( + io::Error::other("log subscriber fell behind").into() + ); + } + self.log_queue.push_back(entry); + } + continue; + }; if revision <= self.revision { return Err(io::Error::new( io::ErrorKind::InvalidData, @@ -290,6 +352,74 @@ mod tests { (address, peer) } + #[test] + fn log_requests_discard_buffered_events_from_the_previous_subscription() { + for follow in [false, true] { + let (address, peer) = websocket_peer(|mut socket, request| { + let notification = Notification { + version: PROTOCOL_VERSION, + event: ImageEvent::Log(runa_protocol::LogEntry { + sequence: 1, + actor_ref: serde_json::from_str("\"SGqJMdr0L5C8dSMmG-Mc2\"").unwrap(), + level: runa_protocol::LogLevel::Info, + message: "old subscription".to_owned(), + }), + }; + socket + .send(Message::Text( + serde_json::to_string(¬ification).unwrap().into(), + )) + .unwrap(); + socket + .send(Message::Text( + serde_json::to_string(&Response { + version: PROTOCOL_VERSION, + request_id: request.request_id, + result: Ok(Reply::Actors(Vec::new())), + }) + .unwrap() + .into(), + )) + .unwrap(); + let request: Request = + serde_json::from_str(socket.read().unwrap().to_text().unwrap()).unwrap(); + socket + .send(Message::Text( + serde_json::to_string(&Response { + version: PROTOCOL_VERSION, + request_id: request.request_id, + result: Ok(Reply::Logs { + entries: Vec::new(), + next_sequence: 0, + has_more: false, + }), + }) + .unwrap() + .into(), + )) + .unwrap(); + assert!(matches!(socket.read().unwrap(), Message::Close(_))); + assert!(matches!( + socket.read(), + Err(tungstenite::Error::ConnectionClosed) + )); + }); + let mut client = WebSocketClient::connect(address).unwrap(); + client.request(ProtocolCommand::ListActors).unwrap(); + assert_eq!(client.log_queue.len(), 1); + client + .request(ProtocolCommand::Logs { + actors: Vec::new(), + follow, + after: 0, + }) + .unwrap(); + assert!(client.log_queue.is_empty()); + client.close().unwrap(); + peer.join().unwrap(); + } + } + #[test] fn websocket_client_completes_close_handshake_with_pending_notifications() { let (address, peer) = websocket_peer(|mut socket, request| { @@ -392,10 +522,10 @@ mod tests { for message in [ serde_json::json!({"version": 2, "request_id": "runa-client-1", "result": {"Ok": {"kind": "actors", "data": []}}}), - serde_json::json!({"version": 4, "request_id": "wrong", + serde_json::json!({"version": 5, "request_id": "wrong", "result": {"Ok": {"kind": "actors", "data": []}}}), - serde_json::json!({"version": 4, "event": {"event": "image_changed", "data": {"revision": 0}}}), - serde_json::json!({"version": 4, "event": {"event": "unknown"}}), + serde_json::json!({"version": 5, "event": {"event": "image_changed", "data": {"revision": 0}}}), + serde_json::json!({"version": 5, "event": {"event": "unknown"}}), ] { let (address, peer) = websocket_peer(move |mut socket, _| { socket diff --git a/crates/runa-core/src/lib.rs b/crates/runa-core/src/lib.rs index 58fc82a..202b28c 100644 --- a/crates/runa-core/src/lib.rs +++ b/crates/runa-core/src/lib.rs @@ -2,7 +2,9 @@ use std::fmt; +mod log; mod value; +pub use log::{LogEntry, LogLevel}; pub use value::{ActorRef, MAX_VALUE_BYTES, MAX_VALUE_DEPTH, MAX_VALUE_NODES, Value}; macro_rules! digest_id { diff --git a/crates/runa-core/src/log.rs b/crates/runa-core/src/log.rs new file mode 100644 index 0000000..827db0c --- /dev/null +++ b/crates/runa-core/src/log.rs @@ -0,0 +1,38 @@ +use serde::{Deserialize, Serialize}; + +use crate::ActorRef; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LogLevel { + Trace, + Debug, + #[default] + Info, + Warn, + Error, + Fatal, +} + +impl LogLevel { + pub fn parse(value: &str) -> Result { + match value { + "trace" => Ok(Self::Trace), + "debug" => Ok(Self::Debug), + "info" => Ok(Self::Info), + "warn" => Ok(Self::Warn), + "error" => Ok(Self::Error), + "fatal" => Ok(Self::Fatal), + _ => Err("log level must be trace, debug, info, warn, error, or fatal".to_owned()), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LogEntry { + pub sequence: u64, + pub actor_ref: ActorRef, + pub level: LogLevel, + pub message: String, +} diff --git a/crates/runa-image-api/src/lib.rs b/crates/runa-image-api/src/lib.rs index 7436aa6..d9e2eae 100644 --- a/crates/runa-image-api/src/lib.rs +++ b/crates/runa-image-api/src/lib.rs @@ -189,6 +189,35 @@ impl ImageApi { self.runtime.inspect(actor) } + pub fn edit_actor( + &mut self, + actor: ActorRef, + state: Option, + messages: Option<(Option, Vec)>, + ) -> RuntimeResult<()> { + self.runtime.edit_actor(actor, state, messages) + } + + pub fn set_log_level( + &mut self, + actor: ActorRef, + level: runa_core::LogLevel, + ) -> RuntimeResult<()> { + self.runtime.set_log_level(actor, level) + } + + pub fn log_level(&self, actor: ActorRef) -> RuntimeResult { + self.runtime.log_level(actor) + } + + pub fn logs(&self, actors: &[ActorRef], after: u64) -> RuntimeResult> { + self.runtime.logs(actors, after) + } + + pub fn set_log_sink(&mut self, sink: impl Fn(runa_core::LogEntry) + 'static) { + self.runtime.set_log_sink(sink); + } + pub fn actors(&self) -> Vec { self.runtime.actors() } diff --git a/crates/runa-protocol/src/lib.rs b/crates/runa-protocol/src/lib.rs index 216c811..90564ab 100644 --- a/crates/runa-protocol/src/lib.rs +++ b/crates/runa-protocol/src/lib.rs @@ -1,9 +1,10 @@ //! Versioned, transport-neutral wire contracts for Runa clients and servers. pub use runa_core::Value; +pub use runa_core::{LogEntry, LogLevel}; use serde::{Deserialize, Serialize}; -pub const PROTOCOL_VERSION: u16 = 4; +pub const PROTOCOL_VERSION: u16 = 5; pub const WEBSOCKET_SUBPROTOCOL: &str = "runa-debug"; /// Coalescible invalidation, not a drain of the image's runtime event queue. @@ -18,6 +19,7 @@ pub struct Notification { #[serde(tag = "event", content = "data", rename_all = "snake_case")] pub enum ImageEvent { ImageChanged { revision: u64 }, + Log(LogEntry), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -44,6 +46,30 @@ impl Request { } match &self.command { + Command::EditActor { + actor_ref, + state, + messages, + } => { + validate_actor_ref(actor_ref)?; + if state.is_none() && messages.is_none() { + return Err(invalid_request("edit must specify state or messages")); + } + if let Some(state) = state { + state.validate().map_err(invalid_request)?; + } + if let Some(messages) = messages { + for message in messages.reserved.iter().chain(&messages.queued) { + message.validate().map_err(invalid_request)?; + } + } + } + Command::SetLogLevel { actor_ref, .. } => validate_actor_ref(actor_ref)?, + Command::Logs { actors, .. } => { + for actor in actors { + validate_actor_ref(actor)?; + } + } Command::ListSources => {} Command::ReadSource { path } | Command::CommitSource { path } @@ -132,6 +158,20 @@ impl Request { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "operation", content = "parameters", rename_all = "snake_case")] pub enum Command { + EditActor { + actor_ref: String, + state: Option, + messages: Option, + }, + SetLogLevel { + actor_ref: String, + level: LogLevel, + }, + Logs { + actors: Vec, + follow: bool, + after: u64, + }, ListSources, ReadSource { path: String, @@ -223,6 +263,13 @@ pub enum Command { }, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MailboxEdit { + pub reserved: Option, + pub queued: Vec, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Response { @@ -234,6 +281,11 @@ pub struct Response { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", content = "data", rename_all = "snake_case")] pub enum Reply { + Logs { + entries: Vec, + next_sequence: u64, + has_more: bool, + }, Sources(Vec), Source(SourceFileDto), SourceCommitted(SourceCommitDto), @@ -241,13 +293,21 @@ pub enum Reply { Acknowledged, Behavior(BehaviorDto), Behaviors(Vec), - BehaviorSource { source: String }, - ActorSpawned { actor_ref: String }, + BehaviorSource { + source: String, + }, + ActorSpawned { + actor_ref: String, + }, Actor(ActorSnapshotDto), Actors(Vec), Events(Vec), - TickCompleted { ticks: usize }, - BehaviorInstalled { implementation_id: String }, + TickCompleted { + ticks: usize, + }, + BehaviorInstalled { + implementation_id: String, + }, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -344,6 +404,7 @@ pub type MessageDto = Value; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ActorSnapshotDto { + pub log_level: LogLevel, pub actor_ref: String, pub mailbox_id: u64, pub implementation_id: String, diff --git a/crates/runa-rune/src/context.rs b/crates/runa-rune/src/context.rs index 5be58de..ae94654 100644 --- a/crates/runa-rune/src/context.rs +++ b/crates/runa-rune/src/context.rs @@ -1,7 +1,8 @@ use std::cell::RefCell; +use std::collections::{BTreeMap, VecDeque}; use std::rc::Rc; -use runa_core::{ActorRef, MAX_VALUE_NODES, Value}; +use runa_core::{ActorRef, LogEntry, LogLevel, MAX_VALUE_NODES, Value}; use rune::runtime::VmResult; use rune::{Any, Value as RuneValue}; @@ -19,6 +20,41 @@ pub struct SendEffect { pub struct ActorContext { actor: ActorRef, effects: Rc>>, + logs: Rc>, +} + +#[derive(Default)] +pub struct ActorLogs { + sequence: u64, + entries: BTreeMap>, + levels: BTreeMap, + sink: Option>, +} + +impl ActorLogs { + pub fn set_sink(&mut self, sink: impl Fn(LogEntry) + 'static) { + self.sink = Some(Box::new(sink)); + } + pub fn level(&self, actor: ActorRef) -> LogLevel { + self.levels.get(&actor).copied().unwrap_or_default() + } + + pub fn set_level(&mut self, actor: ActorRef, level: LogLevel) { + self.levels.insert(actor, level); + } + + pub fn entries(&self, actors: &[ActorRef], after: u64) -> Vec { + let mut entries: Vec<_> = self + .entries + .iter() + .filter(|(actor, _)| actors.is_empty() || actors.contains(actor)) + .flat_map(|(_, entries)| entries.iter()) + .filter(|entry| entry.sequence > after) + .cloned() + .collect(); + entries.sort_by_key(|entry| entry.sequence); + entries + } } impl ActorContext { @@ -26,9 +62,71 @@ impl ActorContext { Self { actor, effects: Rc::default(), + logs: Rc::default(), } } + pub fn with_logs(actor: ActorRef, logs: Rc>) -> Self { + Self { + actor, + effects: Rc::default(), + logs, + } + } + + pub(crate) fn log(&self, level: &str, message: &str) -> VmResult<()> { + let level = match LogLevel::parse(level) { + Ok(level) => level, + Err(error) => return VmResult::panic(error), + }; + let mut logs = self.logs.borrow_mut(); + if level < logs.level(self.actor) { + return VmResult::Ok(()); + } + // Bound each entry so paginated histories fit within the wire limit. + if message.len() > 512 { + return VmResult::panic("log message exceeds 512 UTF-8 bytes"); + } + let Some(sequence) = logs.sequence.checked_add(1) else { + return VmResult::panic("log sequence exhausted"); + }; + logs.sequence = sequence; + let entries = logs.entries.entry(self.actor).or_default(); + if entries.len() == 1000 { + entries.pop_front(); + } + let entry = LogEntry { + sequence, + actor_ref: self.actor, + level, + message: message.to_owned(), + }; + entries.push_back(entry.clone()); + if let Some(sink) = &logs.sink { + sink(entry); + } + VmResult::Ok(()) + } + + pub(crate) fn trace(&self, message: &str) -> VmResult<()> { + self.log("trace", message) + } + pub(crate) fn debug(&self, message: &str) -> VmResult<()> { + self.log("debug", message) + } + pub(crate) fn info(&self, message: &str) -> VmResult<()> { + self.log("info", message) + } + pub(crate) fn warn(&self, message: &str) -> VmResult<()> { + self.log("warn", message) + } + pub(crate) fn error(&self, message: &str) -> VmResult<()> { + self.log("error", message) + } + pub(crate) fn fatal(&self, message: &str) -> VmResult<()> { + self.log("fatal", message) + } + pub fn take_effects(&self) -> Vec { std::mem::take(&mut *self.effects.borrow_mut()) } @@ -80,3 +178,25 @@ impl ActorContext { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn log_validation_is_explicit_and_filtered_messages_are_not_retained() { + let context = ActorContext::new(ActorRef::new()); + assert!(matches!( + context.log("invalid", "message"), + VmResult::Err(_) + )); + let limit = "\u{e9}".repeat(256); + assert!(matches!(context.info(&limit), VmResult::Ok(()))); + let oversized = format!("{limit}x"); + assert!(matches!(context.info(&oversized), VmResult::Err(_))); + assert!(matches!(context.debug(&oversized), VmResult::Ok(()))); + let entries = context.logs.borrow().entries(&[], 0); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].message, limit); + } +} diff --git a/crates/runa-rune/src/lib.rs b/crates/runa-rune/src/lib.rs index ed30ffc..770744d 100644 --- a/crates/runa-rune/src/lib.rs +++ b/crates/runa-rune/src/lib.rs @@ -13,14 +13,14 @@ use rune::termcolor::{ColorChoice, StandardStream}; use rune::{Context, Diagnostics, Module, Source, Sources, Vm}; pub const RUNE_VERSION: &str = "0.14.2"; -pub const COMPILER_CONFIGURATION_ID: &str = "default-modules; native-functions:v2:native_double,native_suspend; runa::ActorReference:v1; runa::ActorContext:v1:send,send_from"; +pub const COMPILER_CONFIGURATION_ID: &str = "default-modules; native-functions:v2:native_double,native_suspend; runa::ActorReference:v1; runa::ActorContext:v2:send,send_from,log,trace,debug,info,warn,error,fatal"; mod execution; pub use execution::{Invocation, InvocationProgress, Slice}; mod value; pub use value::{value_from_rune, value_to_rune}; mod context; -pub use context::{ActorContext, SendEffect}; +pub use context::{ActorContext, ActorLogs, SendEffect}; /// Errors remain visible to the caller; compilation also emits source diagnostics. pub type ExperimentResult = Result>; @@ -66,6 +66,13 @@ impl CompiledProgram { values.ty::()?; values.associated_function("send", ActorContext::send)?; values.associated_function("send_from", ActorContext::send_from)?; + values.associated_function("log", ActorContext::log)?; + values.associated_function("trace", ActorContext::trace)?; + values.associated_function("debug", ActorContext::debug)?; + values.associated_function("info", ActorContext::info)?; + values.associated_function("warn", ActorContext::warn)?; + values.associated_function("error", ActorContext::error)?; + values.associated_function("fatal", ActorContext::fatal)?; values.associated_function( &rune::runtime::Protocol::PARTIAL_EQ, |left: &value::ActorReference, right: &value::ActorReference| left.0 == right.0, diff --git a/crates/runa-runtime/src/lib.rs b/crates/runa-runtime/src/lib.rs index 47c7d29..c51fd87 100644 --- a/crates/runa-runtime/src/lib.rs +++ b/crates/runa-runtime/src/lib.rs @@ -1,7 +1,9 @@ //! Single-worker, in-memory actor runtime. No durability or preemption guarantees. +use std::cell::RefCell; use std::collections::{BTreeMap, VecDeque}; use std::num::NonZeroUsize; +use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Wake, Waker}; @@ -11,9 +13,9 @@ use runa_actor::{ SystemMessage, }; use runa_code::CompiledArtifact; -use runa_core::{ImplementationId, Value}; +use runa_core::{ImplementationId, LogEntry, LogLevel, Value}; use runa_rune::{ - ActorContext, ExperimentResult, Invocation, Slice, value_from_rune, value_to_rune, + ActorContext, ActorLogs, ExperimentResult, Invocation, Slice, value_from_rune, value_to_rune, }; pub type RuntimeResult = ExperimentResult; @@ -49,6 +51,7 @@ pub struct Runtime { implementations: BTreeMap>, next_mailbox: u64, events: Vec, + logs: Rc>, } impl Runtime { @@ -89,6 +92,7 @@ impl Runtime { actor_ref, Actor { snapshot: ActorSnapshot { + log_level: LogLevel::Info, actor_ref, mailbox: MailboxId(self.next_mailbox), implementation, @@ -102,7 +106,7 @@ impl Runtime { mailbox: VecDeque::new(), capacity, active: None, - context: ActorContext::new(actor_ref), + context: ActorContext::with_logs(actor_ref, Rc::clone(&self.logs)), reserved_sender: None, stopping: false, shutdown: false, @@ -158,6 +162,87 @@ impl Runtime { Ok(snapshot) } + pub fn edit_actor( + &mut self, + id: ActorRef, + state: Option, + messages: Option<(Option, Vec)>, + ) -> ExperimentResult<()> { + let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; + if actor.active.is_some() + || !matches!( + actor.snapshot.status, + Status::Paused | Status::Stopped | Status::Failed(_) + ) + { + return Err( + "editing requires a paused, stopped, or failed actor with no active invocation" + .into(), + ); + } + if let Some(state) = &state { + state.validate()?; + } + if let Some((reserved, queued)) = &messages { + for message in reserved.iter().chain(queued) { + message.validate()?; + } + if queued.len() + usize::from(reserved.is_some()) > actor.capacity.get() { + return Err("mailbox full".into()); + } + } + if let Some(state) = state { + actor.snapshot.state = state; + } + let Some((reserved, queued)) = messages else { + return Ok(()); + }; + // Preserve sender metadata for unchanged messages, including the failed reservation. + if actor.snapshot.reserved != reserved { + actor.reserved_sender = None; + } + actor.snapshot.reserved = reserved; + let previous = std::mem::take(&mut actor.mailbox); + actor.mailbox = queued + .into_iter() + .enumerate() + .map(|(index, message)| { + let sender = previous + .get(index) + .filter(|old| old.message == message) + .and_then(|old| old.sender); + Envelope { sender, message } + }) + .collect(); + Ok(()) + } + + pub fn set_log_level(&mut self, id: ActorRef, level: LogLevel) -> ExperimentResult<()> { + self.actors + .get_mut(&id) + .ok_or("unknown actor")? + .snapshot + .log_level = level; + self.logs.borrow_mut().set_level(id, level); + Ok(()) + } + + pub fn log_level(&self, id: ActorRef) -> ExperimentResult { + self.actors.get(&id).ok_or("unknown actor")?; + Ok(self.logs.borrow().level(id)) + } + + pub fn logs(&self, actors: &[ActorRef], after: u64) -> ExperimentResult> { + for actor in actors { + self.actors.get(actor).ok_or("unknown actor")?; + } + Ok(self.logs.borrow().entries(actors, after)) + } + + pub fn set_log_sink(&mut self, sink: impl Fn(LogEntry) + 'static) { + self.logs.borrow_mut().set_sink(sink); + } + pub fn actors(&self) -> Vec { self.actors .values() @@ -444,7 +529,7 @@ impl Runtime { Ok(invocation) => { actor.active = Some(invocation); actor.shutdown = true; - actor.context = ActorContext::new(id); + actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs)); } Err(error) => { self.fail(&id, error.to_string()); @@ -462,7 +547,7 @@ impl Runtime { }; actor.snapshot.reserved = Some(message.message.clone()); actor.reserved_sender = message.sender; - actor.context = ActorContext::new(id); + actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs)); let program = self.implementations[&actor.snapshot.implementation].program(); let invocation = (|| -> ExperimentResult<_> { let state = value_to_rune(&actor.snapshot.state)?; @@ -515,6 +600,136 @@ mod tests { use super::*; use runa_core::MAX_VALUE_NODES; + #[test] + fn editing_is_guarded_atomic_and_can_repair_a_failed_reservation() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let implementation = publish( + &mut runtime, + "pub fn handle(ctx, state, sender, message) { state + message.value }", + )?; + let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?; + assert!( + runtime + .edit_actor(actor, Some(Value::Integer(9)), None) + .is_err() + ); + runtime.send(actor, Value::Null)?; + runtime.send(actor, Value::Integer(4))?; + for _ in 0..10 { + runtime.tick(capacity(32)); + } + assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); + assert!( + runtime + .edit_actor( + actor, + Some(Value::Integer(9)), + Some((Some(Value::Null), vec![Value::Null, Value::Null])) + ) + .is_err() + ); + assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1)); + runtime.edit_actor(actor, Some(Value::Integer(10)), None)?; + assert_eq!(runtime.inspect(actor)?.queued, vec![Value::Integer(4)]); + let message = Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(2))])); + runtime.edit_actor(actor, None, Some((Some(message), vec![])))?; + runtime.restart(actor)?; + for _ in 0..10 { + runtime.tick(capacity(32)); + } + assert_eq!(runtime.inspect(actor)?.state, Value::Integer(12)); + runtime.pause(actor)?; + runtime.edit_actor( + actor, + Some(Value::Integer(20)), + Some((None, vec![Value::Null])), + )?; + assert_eq!(runtime.inspect(actor)?.status, Status::Paused); + runtime.resume(actor)?; + let runaway = publish( + &mut runtime, + "pub fn handle(ctx, state, sender, message) { loop {} }", + )?; + let active = runtime.spawn(runaway, Value::Null, capacity(2))?; + runtime.send(active, Value::Null)?; + runtime.tick(capacity(1)); + runtime.pause(active)?; + assert!( + runtime + .edit_actor(active, Some(Value::Integer(9)), Some((None, vec![]))) + .is_err() + ); + runtime.kill(active)?; + runtime.edit_actor(active, Some(Value::Integer(9)), Some((None, vec![])))?; + Ok(()) + } + + #[test] + fn logs_are_immediate_filtered_ordered_bounded_and_survive_failure() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let emitted = Rc::new(RefCell::new(Vec::new())); + let sink = emitted.clone(); + runtime.set_log_sink(move |entry| sink.borrow_mut().push(entry)); + let implementation = publish( + &mut runtime, + r#"pub fn handle(ctx, state, sender, message) { + ctx.trace("trace"); ctx.debug("debug"); ctx.info("info"); + ctx.warn("warn"); ctx.error("error"); ctx.fatal("fatal"); ctx.log("info", "generic"); + panic("failed after logging") + }"#, + )?; + let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?; + assert_eq!(runtime.log_level(actor)?, LogLevel::Info); + runtime.send(actor, Value::Null)?; + for _ in 0..10 { + runtime.tick(capacity(32)); + } + let entries = runtime.logs(&[actor], 0)?; + assert_eq!( + entries + .iter() + .map(|entry| entry.message.as_str()) + .collect::>(), + ["info", "warn", "error", "fatal", "generic"] + ); + assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); + runtime.set_log_level(actor, LogLevel::Fatal)?; + runtime.restart(actor)?; + for _ in 0..10 { + runtime.tick(capacity(32)); + } + assert_eq!( + runtime + .logs(&[actor], entries.last().unwrap().sequence)? + .len(), + 1 + ); + let busy = publish( + &mut runtime, + r#"pub fn handle(ctx, state, sender, message) { + let n = 0; while n < 1005 { ctx.info("entry"); n += 1; } loop {} + }"#, + )?; + let second = runtime.spawn(busy, Value::Null, capacity(2))?; + runtime.send(second, Value::Null)?; + for _ in 0..100 { + runtime.tick(capacity(1000)); + } + let retained = runtime.logs(&[second], 0)?; + assert_eq!(retained.len(), 1000); + assert!( + retained + .windows(2) + .all(|pair| pair[0].sequence < pair[1].sequence) + ); + runtime.kill(second)?; + assert_eq!(runtime.logs(&[second], 0)?.len(), 1000); + assert_eq!(runtime.logs(&[], 0)?.len(), 1006); + assert_eq!(emitted.borrow().len(), 1011); + assert_eq!(emitted.borrow().last(), retained.last()); + Ok(()) + } + #[test] fn context_sends_copy_at_call_and_are_discarded_on_kill() -> ExperimentResult<()> { let mut runtime = Runtime::default(); diff --git a/crates/runa/src/lib.rs b/crates/runa/src/lib.rs index a7b3b9f..6f18495 100644 --- a/crates/runa/src/lib.rs +++ b/crates/runa/src/lib.rs @@ -65,6 +65,39 @@ impl Server { fn dispatch_command(&mut self, command: Command) -> Result { match command { + Command::EditActor { + actor_ref, + state, + messages, + } => self + .image + .edit_actor( + parse_actor_ref(&actor_ref)?, + state, + messages.map(|messages| (messages.reserved, messages.queued)), + ) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::SetLogLevel { actor_ref, level } => self + .image + .set_log_level(parse_actor_ref(&actor_ref)?, level) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::Logs { actors, after, .. } => { + let actors = actors + .iter() + .map(|actor| parse_actor_ref(actor)) + .collect::, _>>()?; + let mut entries = self.image.logs(&actors, after).map_err(api_error)?; + let has_more = entries.len() > 128; + entries.truncate(128); + let next_sequence = entries.last().map_or(after, |entry| entry.sequence); + Ok(Reply::Logs { + entries, + next_sequence, + has_more, + }) + } Command::ListSources => Ok(Reply::Sources( self.image.sources().into_iter().map(source_dto).collect(), )), @@ -308,6 +341,7 @@ fn behavior_dto(behavior: runa_code::BehaviorSnapshot) -> BehaviorDto { fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto { ActorSnapshotDto { + log_level: snapshot.log_level, actor_ref: snapshot.actor_ref.as_str().to_owned(), mailbox_id: snapshot.mailbox.0, implementation_id: snapshot.implementation.to_string(), @@ -721,7 +755,7 @@ mod tests { assert_eq!( serde_json::to_value(notification).unwrap(), serde_json::json!({ - "version": 4, + "version": 5, "event": {"event": "image_changed", "data": {"revision": 42}} }) ); diff --git a/crates/runa/src/network.rs b/crates/runa/src/network.rs index 4be30dd..bdaa28b 100644 --- a/crates/runa/src/network.rs +++ b/crates/runa/src/network.rs @@ -6,7 +6,7 @@ use futures_util::{SinkExt, StreamExt}; use runa_protocol::{ Command, ImageEvent, Notification, PROTOCOL_VERSION, Request, Response, WEBSOCKET_SUBPROTOCOL, }; -use tokio::sync::{mpsc, oneshot, watch}; +use tokio::sync::{broadcast, mpsc, oneshot, watch}; use tokio::task::JoinSet; use tokio::time::timeout; use tokio_tungstenite::tungstenite::handshake::server::{ @@ -47,6 +47,12 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { let (changes, _) = watch::channel(0_u64); let mut revision = 0_u64; let mut clients = JoinSet::new(); + let (logs, _) = broadcast::channel::(1024); + let log_sender = logs.clone(); + server.image.set_log_sink(move |entry| { + // No queue is kept for absent subscribers; per-actor history remains available. + let _ = log_sender.send(entry); + }); loop { tokio::select! { accepted = listener.accept() => { @@ -57,8 +63,9 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { } let requests = requests.clone(); let changes = changes.subscribe(); + let logs = logs.clone(); clients.spawn(async move { - if let Err(error) = client(stream, requests, changes).await { + if let Err(error) = client(stream, requests, changes, logs).await { eprintln!("runa: client {address}: {error}"); } }); @@ -69,7 +76,7 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { Command::GetBehaviorSource { .. } | Command::InspectActor { .. } | Command::ListActors | Command::GetEvents | Command::ListSources | Command::ReadSource { .. } | Command::SourceHistory { .. } | - Command::RevisionSource { .. }); + Command::RevisionSource { .. } | Command::Logs { .. }); let response = server.dispatch(dispatch.request); let changed = mutates && response.result.is_ok(); // Dispatch may commit even if the initiating client has disconnected. @@ -125,6 +132,7 @@ async fn client( stream: tokio::net::TcpStream, requests: mpsc::Sender, mut changes: watch::Receiver, + logs: broadcast::Sender, ) -> Result<(), Box> { let config = WebSocketConfig::default() .max_message_size(Some(MAX_MESSAGE_BYTES)) @@ -135,15 +143,29 @@ async fn client( tokio_tungstenite::accept_hdr_async_with_config(stream, DebugProtocol, Some(config)), ) .await??; + let mut subscription: Option> = None; + let mut log_filter = Vec::::new(); + let mut log_cursor = 0; loop { tokio::select! { message = socket.next() => { match message { Some(Ok(Message::Text(text))) => { let request: Request = serde_json::from_str(&text)?; + let requested_logs = match &request.command { + Command::Logs { actors, follow, .. } => Some((actors.clone(), *follow)), + _ => None, + }; + let receiver = requested_logs.as_ref().filter(|(_, follow)| *follow).map(|_| logs.subscribe()); let (respond, response) = oneshot::channel(); requests.send(Dispatch { request, respond }).await?; - let encoded = serde_json::to_string(&response.await?)?; + let response = response.await?; + if let (Some((actors, _)), Ok(runa_protocol::Reply::Logs { next_sequence, .. })) = (requested_logs, &response.result) { + subscription = receiver; + log_filter = actors; + log_cursor = *next_sequence; + } + let encoded = serde_json::to_string(&response)?; if encoded.len() > MAX_MESSAGE_BYTES { return Err("WebSocket response exceeds 1 MiB".into()); } @@ -172,6 +194,19 @@ async fn client( let text = serde_json::to_string(¬ification)?; timeout(WRITE_TIMEOUT, socket.send(Message::Text(text.into()))).await??; } + entry = async { + match &mut subscription { + Some(receiver) => receiver.recv().await, + None => std::future::pending().await, + } + } => { + let entry = entry?; + if entry.sequence > log_cursor && (log_filter.is_empty() || log_filter.iter().any(|actor| actor == entry.actor_ref.as_str())) { + log_cursor = entry.sequence; + let text = serde_json::to_string(&Notification { version: PROTOCOL_VERSION, event: ImageEvent::Log(entry) })?; + timeout(WRITE_TIMEOUT, socket.send(Message::Text(text.into()))).await??; + } + } } } } diff --git a/crates/runa/tests/local_client.rs b/crates/runa/tests/local_client.rs index 86a292d..0afe501 100644 --- a/crates/runa/tests/local_client.rs +++ b/crates/runa/tests/local_client.rs @@ -86,7 +86,9 @@ fn revision(socket: &mut tungstenite::WebSocket) -> u64 { let event: runa_protocol::Notification = serde_json::from_str(message.to_text().unwrap()).unwrap(); assert_eq!(event.version, runa_protocol::PROTOCOL_VERSION); - let runa_protocol::ImageEvent::ImageChanged { revision } = event.event; + let runa_protocol::ImageEvent::ImageChanged { revision } = event.event else { + panic!("expected image invalidation") + }; revision } @@ -178,7 +180,7 @@ fn websocket_broadcasts_non_destructive_invalidations_for_mailbox_state_and_life first .send(Message::Text( serde_json::to_string(&runa_protocol::Request { - version: 4, + version: 5, request_id: "read-fence".to_owned(), command: Command::ListActors, }) diff --git a/docs/protocol.md b/docs/protocol.md index 4fd780c..25df560 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -11,7 +11,7 @@ trusted local development. Both the REPL and VS Code use WebSockets. ## Version 4 -Every request and response carries `version: 4`. Requests also carry a +Every request and response carries `version: 5`. Requests also carry a non-empty `request_id` used to correlate the response. A server must reject unsupported versions with `unsupported_version` rather than interpreting a request using a different contract. Events carry the protocol version and do @@ -21,7 +21,7 @@ Requests have a tagged command object: ```json { - "version": 4, + "version": 5, "request_id": "req-1", "command": { "operation": "send", @@ -40,7 +40,7 @@ Requests have a tagged command object: } ``` -The version 4 operations are source listing/read/write, commit/history/revision +The version 5 operations are source listing/read/write, commit/history/revision retrieval/rollback, actor spawn, send, tick, inspect, actor listing, event retrieval, privileged system messages, and cancellation. Migration closures are intentionally excluded: they are host callbacks, not @@ -155,6 +155,32 @@ the actor tree and open snapshot documents. Manual refresh remains available. ## WebSocket transport and notifications +### Actor editing and log subscriptions (v5) + +- `edit_actor {actor_ref,state,messages}`: `state` is a tagged portable value or + null to leave it unchanged. `messages` is null to leave the mailbox unchanged, + or `{reserved,queued}` where reserved is a tagged value or null for no + reservation, and queued is an ordered array of tagged values. At least one + edit is required. Only paused/stopped/failed actors with no active execution + are editable; values and mailbox capacity are validated atomically. +- `set_log_level {actor_ref,level}`: level is one of `trace`, `debug`, `info`, + `warn`, `error`, `fatal`. Both mutations return `acknowledged` and invalidate + image views. Actor snapshots include `log_level`. +- `logs {actors,follow,after}`: actors is an array of references (empty means + all); after is an exclusive u64 sequence cursor. Reply kind `logs` carries + `{entries,next_sequence,has_more}`, with at most 128 entries per page. + Each entry is `{sequence,actor_ref,level,message}`. Missing actors are errors. + History queries are read-only and do not invalidate views. +- With follow true, the connection registers the filter before history is read. + Subsequent notification envelopes carry + `{event:"log",data:{sequence,actor_ref,level,message}}`. Reissuing `logs` + replaces the filter; follow false unsubscribes. Matching entries newer than + the reply's cursor are emitted after its reply, allowing history/follow + deduplication by sequence. Other connections receive no log events. + Transport and client buffers are bounded at 1,024 entries; lag is an explicit + connection error, not a guaranteed-delivery stream. Disconnect removes the + subscription. Image invalidations remain coalescible and independent of logs. + On `quit`, `exit`, Ctrl+C, or end-of-input, the CLI completes the WebSocket close handshake with a 5-second deadline before releasing the connection. A failed handshake is reported rather than treating an abrupt reset as a normal close. @@ -162,7 +188,7 @@ handshake is reported rather than treating an abrupt reset as a normal close. 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 is version 4. +listener. The request/response contract is version 5. 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 @@ -179,7 +205,7 @@ Every upgraded connection automatically receives non-destructive invalidations: ```json { - "version": 4, + "version": 5, "event": { "event": "image_changed", "data": { "revision": 42 } diff --git a/docs/runtime.md b/docs/runtime.md index 96ae757..ffcc57b 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -1,5 +1,38 @@ # Phase 1 actor prototype +## Actor repair and logging + +`set-state ` replaces committed state without touching the mailbox. +`set-messages {"reserved":[],"queued":[]}` replaces the reserved and queued +messages without changing state. The reserved array has zero or one element +(so a null-valued message is distinguishable from no reservation). +Both operations require paused, stopped, or failed status and no active invocation. +Idle actors must be paused first; a paused active invocation must be killed first. +All value and capacity checks finish before any changes are applied. Edits do +not resume/restart an actor or change its implementation/incarnation. +Unchanged messages at unchanged positions retain sender metadata; replaced +messages are host-originated. Removing the failed reservation avoids retrying it. + +`ctx.log(level, message)` accepts a lowercase severity string and message string. +`ctx.trace`, `ctx.debug`, `ctx.info`, `ctx.warn`, `ctx.error`, and `ctx.fatal` +delegate to it. Trace < debug < info < warn < error < fatal; default info. +`log-level ` changes the minimum accepted severity, including +during suspended execution. Lower-severity entries are neither stored nor sent. +Accepted messages longer than 512 UTF-8 bytes fail explicitly. Logs use an +image-wide increasing sequence and include actor identity, level, and message. +Each actor retains its last 1,000 accepted entries across restart, in memory only. +Logging is immediate and does not roll back if a handler fails, is cancelled, or +is killed. This is diagnostic output, not a transactional actor send. + +`logs [-f|--follow] [actor...]` is available in both CLI modes. REPL names resolve +to references before the request; an empty filter means all actors. History is +read non-destructively in pages, preserving sequence order. Follow requests +register a per-connection server-side filter and receive only matching future +log events; unsubscribed connections receive no log events. Ctrl+C unsubscribes, +returning to the REPL or gracefully closing the one-shot connection. Live log +buffers are bounded; lagging subscriptions fail explicitly rather than silently +losing entries. History still only contains the retained tail, not durable logs. + Start the server-owned, in-memory image: ```sh @@ -116,7 +149,7 @@ compilation and actor installation; invalid Rune source can be committed. Commit IDs are generated from a readable lowercase alphabet, not caller-chosen revision numbers. Source and history remain in memory until server exit. The older named behavior registry survives as an internal compatibility helper, -but its publish/history/install commands are not part of the v4 wire contract. +but its publish/history/install commands are not part of the v5 wire contract. The runtime's low-level `install` operation switches an actor's implementation only when it is idle, has no active invocation, and has no reserved input. Queued diff --git a/vscode/README.md b/vscode/README.md index ce9bdc7..1100e7d 100644 --- a/vscode/README.md +++ b/vscode/README.md @@ -1,7 +1,7 @@ # Runa VS Code client This is a development client for a trusted localhost Runa image. It uses -the [version 4 protocol](../docs/protocol.md) over WebSockets with the `runa-debug` +the [version 5 protocol](../docs/protocol.md) over WebSockets with the `runa-debug` subprotocol, not Rust APIs, and never hosts actors or starts its own image. @@ -98,6 +98,15 @@ in the Runa output channel and an error notification; commands are never retried ## Image source files +Actor context menus also offer **Edit Actor State...** and +**Edit Reserved and Queued Messages...**. They prefill JSON from the current +snapshot; editing requires a paused, stopped, or failed actor with no active +execution. Message input uses `{"reserved":[],"queued":[]}` with zero or one +reserved message. Clearing the reservation removes the failed input that would +otherwise retry on restart. Edits do not resume actors. **Set Actor Log Level...** +selects trace/debug/info/warn/error/fatal (default info); snapshots show the level. +Use CLI `logs [-f] [actor...]` for retained/live log output. + The dedicated **Runa** Activity Bar sidebar contains **Image Sources** first, followed by **Actors**. Both views are visible by default and do not require an open workspace. The source tree uses native VS Code tree items, file icons, diff --git a/vscode/integration/host.ts b/vscode/integration/host.ts index 2352ae5..537eedf 100644 --- a/vscode/integration/host.ts +++ b/vscode/integration/host.ts @@ -114,7 +114,7 @@ export async function run(): Promise { message: { type: "string", data: "queued from another client" }, }, }), - { version: 4, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, + { version: 5, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, ); await refreshed; assert.match(document.getText(), /queued from another client/); @@ -146,6 +146,28 @@ export async function run(): Promise { assert.match(document.getText(), /queued from another client/); const old = await vscode.workspace.openTextDocument(vscode.Uri.parse(firstUri)); assert.match(old.getText(), /"connected": false/); + await vscode.commands.executeCommand("runa.editState", { actor_ref: actorRef }, "77"); + await vscode.commands.executeCommand( + "runa.editMessages", + { actor_ref: actorRef }, + '{"reserved":[],"queued":[{"op":"add","value":2}]}', + ); + const recovered = documentChanged(document, (text) => { + const actor = JSON.parse(text).actor; + return ( + actor.state.data === 79 && actor.log_level === "warn" && actor.status.status === "idle" + ); + }); + await vscode.commands.executeCommand("runa.logLevel", { actor_ref: actorRef }, "warn"); + await vscode.commands.executeCommand("runa.restart", { actor_ref: actorRef }); + await vscode.commands.executeCommand("runa.tick", 100); + await recovered; + await vscode.commands.executeCommand("runa.inspect", { actor_ref: actorRef }); + const repaired = vscode.window.activeTextEditor?.document; + assert.ok(repaired); + assert.equal(JSON.parse(repaired.getText()).actor.state.data, 79, repaired.getText()); + assert.equal(JSON.parse(repaired.getText()).actor.log_level, "warn"); + assert.equal(JSON.parse(repaired.getText()).actor.status.status, "idle"); console.log( "Runa extension host: spawn, send, system messages, ticks, inspection, automatic refresh, reconnect passed", ); diff --git a/vscode/integration/logs.test.ts b/vscode/integration/logs.test.ts new file mode 100644 index 0000000..bf78096 --- /dev/null +++ b/vscode/integration/logs.test.ts @@ -0,0 +1,237 @@ +import assert from "node:assert/strict"; +import { execFile, spawn } from "node:child_process"; +import { once } from "node:events"; +import { resolve } from "node:path"; +import test from "node:test"; +import { promisify } from "node:util"; +import WebSocket from "ws"; +import { ImageClient } from "../src/client"; +import { parseValue } from "../src/values"; +import { control, startImage } from "./server"; + +test( + "actor recovery edits, log filtering, history and CLI follow share the real image", + { timeout: 15000 }, + async (t) => { + const image = await startImage(); + const client = await ImageClient.connect(image.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await client.close(); + await image.stop(); + }); + await client.writeSource( + "actors/logger", + null, + `pub fn handle(ctx, state, sender, message) { + ctx.trace("trace"); ctx.info(message); ctx.error("error"); state + }`, + ); + const first = await client.spawn("actors/logger", parseValue("0")); + const second = await client.spawn("actors/logger", parseValue("0")); + await assert.rejects(client.editActor(first, parseValue("5"), null), /editing requires/); + await client.system(first, { kind: "pause" }); + await client.editActor(first, parseValue("5"), { + reserved: parseValue('"repaired"'), + queued: [parseValue('"queued"')], + }); + assert.deepEqual((await client.inspectActor(first)).state, parseValue("5")); + await client.system(first, { kind: "kill" }); + await client.system(first, { kind: "restart" }); + await client.tick(100); + const executable = resolve(__dirname, "../../../target/debug/runa-cli"); + const execute = promisify(execFile); + const run = async (...args: string[]) => + (await execute(executable, ["--connect", image.address, ...args])).stdout; + assert.match(await run("logs", first), /repaired/); + assert.match(await run("logs", first), /queued/); + assert.doesNotMatch(await run("logs", first), /Trace/); + await run("log-level", first, "error"); + await client.send(first, parseValue('"filtered"')); + await client.tick(100); + assert.doesNotMatch(await run("logs", first), /filtered/); + await client.setLogLevel(first, "trace"); + await client.send(first, parseValue('"visible"')); + await client.tick(100); + assert.match(await run("logs", first), /Trace/); + + const following = spawn(executable, ["--connect", image.address, "logs", "-f", first], { + stdio: ["ignore", "pipe", "pipe"], + }); + t.after(() => { + if (following.exitCode === null) following.kill(); + }); + let output = ""; + let stderr = ""; + let ready = () => {}; + const history = new Promise((resolve) => { + ready = resolve; + }); + let future = () => {}; + const received = new Promise((resolve) => { + future = resolve; + }); + following.stdout.on("data", (chunk: Buffer) => { + output += chunk.toString(); + if (output.includes("visible")) ready(); + if (output.includes("future")) future(); + }); + following.stderr.on("data", (chunk: Buffer) => { + stderr += chunk.toString(); + }); + await history; + await client.send(second, parseValue('"not-subscribed"')); + await client.send(first, parseValue('"future"')); + await client.tick(100); + await received; + const exited = once(following, "exit"); + following.kill("SIGINT"); + const [code] = await exited; + assert.equal(code, 0, stderr); + assert.equal(stderr, ""); + assert.doesNotMatch(output, /not-subscribed/); + assert.match(await run("logs"), /not-subscribed/); + const repl = spawn(executable, ["--connect", image.address, "repl"], { + stdio: ["pipe", "pipe", "pipe"], + }); + t.after(() => { + if (repl.exitCode === null) repl.kill(); + }); + let replOutput = ""; + let replError = ""; + let followingName = () => {}; + const namedHistory = new Promise((resolve) => { + followingName = resolve; + }); + repl.stdout.on("data", (chunk: Buffer) => { + replOutput += chunk.toString(); + if (replOutput.includes("Info: named-filter")) followingName(); + }); + repl.stderr.on("data", (chunk: Buffer) => { + replError += chunk.toString(); + }); + repl.stdin.write( + 'let watched = spawn actors/logger 0\nsend watched "named-filter"\ntick 100\nlogs -f watched\n', + ); + await namedHistory; + const replExit = once(repl, "exit"); + repl.kill("SIGINT"); + repl.stdin.end("ps\nquit\n"); + const [replCode] = await replExit; + assert.equal(replCode, 0, replError); + assert.doesNotMatch(replOutput, /^error:/im); + assert.equal(replError, ""); + await client.system(first, { kind: "kill" }); + await run("set-state", first, "19"); + await run("set-messages", first, '{"reserved":[],"queued":["fixed"]}'); + assert.deepEqual((await client.inspectActor(first)).state, parseValue("19")); + assert.deepEqual((await client.inspectActor(first)).queued, [parseValue('"fixed"')]); + assert.doesNotMatch(image.diagnostics(), /reset without closing handshake/i); + }, +); + +test( + "server sends log events only to subscribed clients and applies actor filters", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + const client = await ImageClient.connect(image.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await client.close(); + await image.stop(); + }); + await client.writeSource( + "actors/logger", + null, + 'pub fn handle(ctx, state, sender, message) { ctx.info("hello"); state }', + ); + const first = await client.spawn("actors/logger", parseValue("0")); + const second = await client.spawn("actors/logger", parseValue("0")); + const socket = new WebSocket(`ws://${image.address}/`, "runa-debug"); + t.after(() => socket.close()); + await once(socket, "open"); + const entries: { actor_ref: string }[] = []; + let reply = () => {}; + const subscribed = new Promise((resolve) => { + reply = resolve; + }); + let event = () => {}; + const logged = new Promise((resolve) => { + event = resolve; + }); + socket.on("message", (data) => { + const message = JSON.parse(data.toString()); + if (message.request_id === "subscribe") reply(); + if (message.event?.event === "log") { + entries.push(message.event.data); + event(); + } + }); + socket.send( + JSON.stringify({ + version: 5, + request_id: "subscribe", + command: { operation: "logs", parameters: { actors: [first], follow: true, after: 0 } }, + }), + ); + await subscribed; + await client.send(second, parseValue("null")); + await client.send(first, parseValue("null")); + await client.tick(100); + await logged; + assert.deepEqual( + entries.map((entry) => entry.actor_ref), + [first], + ); + const history = JSON.stringify( + await control(image.address, { + operation: "logs", + parameters: { actors: [second], follow: false, after: 0 }, + }), + ); + assert.ok(history.includes(second)); + assert.ok(!history.includes(first)); + }, +); + +test( + "log history pagination preserves ordered entries beyond a single page", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + const client = await ImageClient.connect(image.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await client.close(); + await image.stop(); + }); + await client.writeSource( + "actors/logger", + null, + `pub fn handle(ctx, state, sender, message) { + let n = 0; while n < 300 { ctx.info("line"); n += 1; } state + }`, + ); + const actor = await client.spawn("actors/logger", parseValue("0")); + await client.send(actor, parseValue("null")); + await client.tick(1000); + const executable = resolve(__dirname, "../../../target/debug/runa-cli"); + const { stdout } = await promisify(execFile)(executable, [ + "--connect", + image.address, + "logs", + actor, + ]); + const lines = stdout.trim().split("\n"); + assert.equal(lines.length, 300); + const sequences = lines.map((line) => Number(line.split(" ")[0])); + assert.deepEqual( + sequences, + Array.from({ length: 300 }, (_, index) => index + 1), + ); + }, +); diff --git a/vscode/integration/server.test.ts b/vscode/integration/server.test.ts index 6a4d5ad..1b3a074 100644 --- a/vscode/integration/server.test.ts +++ b/vscode/integration/server.test.ts @@ -38,7 +38,7 @@ test( operation: "send", parameters: { actor_ref: actorRef, message: payload }, }), - { version: 4, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, + { version: 5, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, ); const failures: Error[] = []; @@ -74,7 +74,7 @@ test( // Reading from the extension must not drain the shared runtime event queue. const events = await control(image.address, { operation: "get_events" }); assert.deepEqual(events, { - version: 4, + version: 5, request_id: "integration-control", result: { Ok: { diff --git a/vscode/integration/server.ts b/vscode/integration/server.ts index 5ddcace..3a67985 100644 --- a/vscode/integration/server.ts +++ b/vscode/integration/server.ts @@ -90,7 +90,7 @@ export async function control(address: string, command: unknown): Promise reject(new Error("Control connection closed"))); socket.once("error", reject); }); - socket.send(json({ version: 4, request_id: "integration-control", command })); + socket.send(json({ version: 5, request_id: "integration-control", command })); return await response; } finally { clearTimeout(timer); diff --git a/vscode/integration/websocket.test.ts b/vscode/integration/websocket.test.ts index fd66cbd..b9ddd05 100644 --- a/vscode/integration/websocket.test.ts +++ b/vscode/integration/websocket.test.ts @@ -17,7 +17,7 @@ test( await once(socket, "open"); assert.equal(socket.protocol, WEBSOCKET_SUBPROTOCOL); const request = json({ - version: 4, + version: 5, request_id: "fragmented", command: { operation: "list_actors" }, }); @@ -27,7 +27,7 @@ test( socket.send(text.slice(100), { fin: true }); const [response] = await reply; assert.deepEqual(JSON.parse(String(response)), { - version: 4, + version: 5, request_id: "fragmented", result: { Ok: { kind: "actors", data: [] } }, }); @@ -52,7 +52,7 @@ test( socket.send(" ", { fin: true }); await closed; assert.deepEqual(await control(image.address, { operation: "list_actors" }), { - version: 4, + version: 5, request_id: "integration-control", result: { Ok: { kind: "actors", data: [] } }, }); diff --git a/vscode/package.json b/vscode/package.json index 3477b17..fef2472 100644 --- a/vscode/package.json +++ b/vscode/package.json @@ -39,6 +39,24 @@ }, "contributes": { "commands": [ + { + "command": "runa.editState", + "title": "Edit Actor State...", + "category": "Runa", + "enablement": "runa.connected" + }, + { + "command": "runa.editMessages", + "title": "Edit Reserved and Queued Messages...", + "category": "Runa", + "enablement": "runa.connected" + }, + { + "command": "runa.logLevel", + "title": "Set Actor Log Level...", + "category": "Runa", + "enablement": "runa.connected" + }, { "command": "runa.connect", "title": "Connect to Local Image", @@ -221,6 +239,21 @@ } ], "view/item/context": [ + { + "command": "runa.editState", + "when": "view == runa.actors && viewItem == runa.actor", + "group": "edit@1" + }, + { + "command": "runa.editMessages", + "when": "view == runa.actors && viewItem == runa.actor", + "group": "edit@2" + }, + { + "command": "runa.logLevel", + "when": "view == runa.actors && viewItem == runa.actor", + "group": "logging@1" + }, { "command": "runa.spawnSource", "when": "view == runa.sources && viewItem == runa.sourceFile && runa.connected", diff --git a/vscode/src/client.ts b/vscode/src/client.ts index 821b95e..80e78bd 100644 --- a/vscode/src/client.ts +++ b/vscode/src/client.ts @@ -20,6 +20,8 @@ import { SystemMessage, Value, WEBSOCKET_SUBPROTOCOL, + LogLevel, + LOG_LEVELS, } from "./protocol"; export function parseEndpoint(endpoint: string): { host: string; port: number } { @@ -221,6 +223,34 @@ export class ImageClient { }); } + async setLogLevel(actorRef: string, level: LogLevel): Promise { + if (!isActorRef(actorRef) || !LOG_LEVELS.includes(level)) + throw new Error("Invalid actor or log level"); + await this.request({ operation: "set_log_level", parameters: { actor_ref: actorRef, level } }); + } + + async editActor( + actorRef: string, + state: Value | null, + messages: { reserved: Value | null; queued: Value[] } | null, + ): Promise { + if (!isActorRef(actorRef)) throw new Error("Invalid actor reference"); + await this.request({ + operation: "edit_actor", + parameters: { + actor_ref: actorRef, + state: state === null ? null : decodeValue(state), + messages: + messages === null + ? null + : { + reserved: messages.reserved === null ? null : decodeValue(messages.reserved), + queued: messages.queued.map(decodeValue), + }, + }, + }); + } + async system(actorRef: string, message: SystemMessage): Promise { if (!isActorRef(actorRef)) throw new Error("Invalid actor reference"); const kinds = ["stop", "kill", "restart", "pause", "resume", "reload", "become"]; diff --git a/vscode/src/extension.ts b/vscode/src/extension.ts index f716200..19e324f 100644 --- a/vscode/src/extension.ts +++ b/vscode/src/extension.ts @@ -1,8 +1,16 @@ import * as vscode from "vscode"; import { ImageClient, parseEndpoint } from "./client"; -import { ActorSnapshot, formatSnapshot, isSourcePath, SystemKind, SystemMessage } from "./protocol"; +import { + ActorSnapshot, + formatSnapshot, + isSourcePath, + SystemKind, + SystemMessage, + LOG_LEVELS, + LogLevel, +} from "./protocol"; import { RefreshQueue } from "./refresh"; -import { parseValue } from "./values"; +import { parseValue, formatValue } from "./values"; import { ImageSourceFileSystem, ImageSourceTree, @@ -279,7 +287,7 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS view.message = tree.actors.length ? undefined : "No actors. Spawn a .rn file from the source folder."; - output.appendLine(`Connected to ${address} (protocol v4)`); + output.appendLine(`Connected to ${address} (protocol v5)`); }; const inspect = async (selected?: ActorSnapshot) => { @@ -338,9 +346,10 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS }; }; - const jsonInput = (title: string) => + const jsonInput = (title: string, value?: string) => vscode.window.showInputBox({ title, + value, prompt: 'JSON value; use {"$actor":"@reference"} for an actor reference', validateInput: (source) => { try { @@ -548,6 +557,82 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS void vscode.window.showInformationMessage(`Runa: Message queued for @${actorRef}`); }; + const logLevel = async (selected?: { actor_ref: string }, level?: LogLevel) => { + const connection = active(); + const actorRef = await chooseActor(selected); + if (actorRef === undefined) return; + const chosen = + typeof level === "string" + ? level + : await vscode.window.showQuickPick(LOG_LEVELS, { + title: `Minimum log level for @${actorRef}`, + }); + if (chosen === undefined) return; + const checked = LOG_LEVELS.find((level) => level === chosen); + if (checked === undefined) throw new Error("Invalid log level"); + connection.check(); + await connection.client.setLogLevel(actorRef, checked); + output.appendLine(`Log level for @${actorRef}: ${chosen}`); + }; + + const editActor = async ( + kind: "state" | "messages", + selected?: { actor_ref: string }, + source?: string, + ) => { + const connection = active(); + const actorRef = await chooseActor(selected); + if (actorRef === undefined) return; + const actor = await connection.client.inspectActor(actorRef); + if ( + actor.execution !== null || + !["paused", "stopped", "failed"].includes(actor.status.status) + ) { + throw new Error( + "Editing requires a paused, stopped, or failed actor with no active invocation", + ); + } + const current = + kind === "state" + ? actor.state + : { + type: "record" as const, + data: { + reserved: { + type: "list" as const, + data: actor.reserved === null ? [] : [actor.reserved], + }, + queued: { type: "list" as const, data: actor.queued }, + }, + }; + const input = + typeof source === "string" + ? source + : await jsonInput(`Edit ${kind} for @${actorRef}`, formatValue(current)); + if (input === undefined) return; + const value = parseValue(input); + connection.check(); + if (kind === "state") { + await connection.client.editActor(actorRef, value, null); + } else { + if ( + value.type !== "record" || + Object.keys(value.data).length !== 2 || + value.data.reserved?.type !== "list" || + value.data.queued?.type !== "list" || + value.data.reserved.data.length > 1 + ) { + throw new Error('Use {"reserved":[],"queued":[]} with zero or one reserved message'); + } + await connection.client.editActor(actorRef, null, { + reserved: value.data.reserved.data[0] ?? null, + queued: value.data.queued.data, + }); + } + await refresh(); + output.appendLine(`Updated ${kind} for @${actorRef}`); + }; + const system = async (kind: SystemKind, selected?: { actor_ref: string }) => { const connection = active(); const actorRef = await chooseActor(selected); @@ -639,6 +724,13 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS command("runa.sourceHistory", sourceHistory), command("runa.rollbackSource", rollbackSource), command("runa.send", send), + command("runa.logLevel", logLevel), + command("runa.editState", (selected?: { actor_ref: string }, source?: string) => + editActor("state", selected, source), + ), + command("runa.editMessages", (selected?: { actor_ref: string }, source?: string) => + editActor("messages", selected, source), + ), systemCommand("stop"), systemCommand("kill"), systemCommand("restart"), diff --git a/vscode/src/protocol.ts b/vscode/src/protocol.ts index 869e802..8f5b13f 100644 --- a/vscode/src/protocol.ts +++ b/vscode/src/protocol.ts @@ -1,6 +1,6 @@ import { parse, stringify } from "lossless-json"; -export const PROTOCOL_VERSION = 4; +export const PROTOCOL_VERSION = 5; export const MAX_MESSAGE_BYTES = 1024 * 1024; export const WEBSOCKET_SUBPROTOCOL = "runa-debug"; const ACTOR_REF = /^[A-Za-z0-9_-]{21}$/; @@ -14,6 +14,7 @@ export type Value = | { type: "record"; data: Record }; export interface ActorSnapshot { + log_level: LogLevel; actor_ref: string; mailbox_id: bigint; implementation_id: string; @@ -34,6 +35,8 @@ export interface ActorSnapshot { export type SystemKind = "stop" | "kill" | "restart" | "pause" | "resume" | "reload"; export type SystemMessage = { kind: SystemKind } | { kind: "become"; path: string }; +export const LOG_LEVELS = ["trace", "debug", "info", "warn", "error", "fatal"] as const; +export type LogLevel = (typeof LOG_LEVELS)[number]; export interface SourceFile { path: string; @@ -47,6 +50,15 @@ export interface SourceCommit { } export type Command = + | { operation: "set_log_level"; parameters: { actor_ref: string; level: LogLevel } } + | { + operation: "edit_actor"; + parameters: { + actor_ref: string; + state: Value | null; + messages: { reserved: Value | null; queued: Value[] } | null; + }; + } | { operation: "list_actors" | "list_sources" } | { operation: "read_source" | "commit_source" | "source_history"; parameters: { path: string } } | { @@ -189,7 +201,10 @@ function decodeActor(input: unknown): ActorSnapshot { throw new Error("Unknown actor status"); } const execution = actor.execution === null ? null : object(actor.execution); + const level = LOG_LEVELS.find((level) => level === actor.log_level); + if (level === undefined) throw new Error("Invalid actor log level"); return { + log_level: level, actor_ref: actor.actor_ref, mailbox_id: uint(actor.mailbox_id), implementation_id: text(actor.implementation_id), @@ -328,7 +343,10 @@ export function decodeResponse(line: string, requestId: string, command: Command return { kind: "tick_completed", data: { ticks } }; } if ( - (command.operation === "send" || command.operation === "system") && + (command.operation === "send" || + command.operation === "system" || + command.operation === "set_log_level" || + command.operation === "edit_actor") && reply.kind === "acknowledged" ) return { kind: "acknowledged" }; diff --git a/vscode/src/values.ts b/vscode/src/values.ts index 8636a42..6fa226f 100644 --- a/vscode/src/values.ts +++ b/vscode/src/values.ts @@ -1,6 +1,34 @@ -import { parse } from "lossless-json"; +import { parse, stringify } from "lossless-json"; import { decodeValue, isActorRef, MAX_MESSAGE_BYTES, Value } from "./protocol"; +export function formatValue(value: Value): string { + const natural = (value: Value): unknown => { + switch (value.type) { + case "null": + return null; + case "bool": + case "integer": + case "string": + return value.data; + case "actor_ref": + return { $actor: value.data }; + case "list": + return value.data.map(natural); + case "record": { + const record = Object.fromEntries( + Object.entries(value.data).map(([key, child]) => [key, natural(child)]), + ); + return Object.keys(record).length === 1 && ("$actor" in record || "$record" in record) + ? { $record: record } + : record; + } + } + }; + const result = stringify(natural(value)); + if (result === undefined) throw new Error("Could not encode actor value"); + return result; +} + /** Natural JSON uses the same $actor/$record marker syntax as the REPL. */ export function parseValue(source: string): Value { if (Buffer.byteLength(source) > MAX_MESSAGE_BYTES) throw new Error("JSON input exceeds 1 MiB"); diff --git a/vscode/test/client.test.ts b/vscode/test/client.test.ts index 5108c60..4170e26 100644 --- a/vscode/test/client.test.ts +++ b/vscode/test/client.test.ts @@ -49,7 +49,7 @@ test("fragmented UTF-8 replies and simultaneous reads preserve correlation and v const operations: string[] = []; const address = await fakeImage(t, (request, socket) => { operations.push(request.command.operation); - assert.equal(request.version, 4); + assert.equal(request.version, 5); const reply = request.command.operation === "list_actors" ? { kind: "actors", data: [actor()] } @@ -74,7 +74,7 @@ test("protocol errors preserve the connection and are never retried", async (t) socket.send( requests === 1 ? json({ - version: 4, + version: 5, request_id: request.request_id, result: { Err: { code: "not_found", message: "Missing actor" } }, }) @@ -99,7 +99,7 @@ test("protocol errors preserve the connection and are never retried", async (t) t.after(() => client.close()); const overhead = Buffer.byteLength( json({ - version: 4, + version: 5, request_id: "runa-vscode-1", command: { operation: "send", @@ -162,7 +162,7 @@ test("bad correlation, malformed JSON/UTF-8, oversize, truncation, and wrong ver socket.send('{"version":3', { fin: false }); socket.terminate(); } else if (mode === "binary") socket.send(Buffer.from("{}")); - else if (mode === "event") socket.send(json({ version: 4, event: { event: "unknown" } })); + else if (mode === "event") socket.send(json({ version: 5, event: { event: "unknown" } })); else socket.send( json({ @@ -315,7 +315,7 @@ test("failed connection reports the error", async () => { ); }); -test("source and system commands use the exact v4 wire contract", async (t) => { +test("source and system commands use the exact v5 wire contract", async (t) => { const requests: unknown[] = []; const file = { path: "actors/counter", source_id: "source:2", source: "pub fn handle() {}\n" }; const address = await fakeImage(t, (request, socket) => { diff --git a/vscode/test/fixtures.ts b/vscode/test/fixtures.ts index d51001e..2a56950 100644 --- a/vscode/test/fixtures.ts +++ b/vscode/test/fixtures.ts @@ -8,6 +8,7 @@ export const REF = "abcdefghijklmnopqrstu"; export function actor(): ActorSnapshot { return { + log_level: "info", actor_ref: REF, mailbox_id: 18446744073709551615n, implementation_id: "implementation:test", @@ -42,7 +43,7 @@ export interface TestRequest { } export function response(request: TestRequest, reply: unknown): string { - return json({ version: 4, request_id: request.request_id, result: { Ok: reply } }); + return json({ version: 5, request_id: request.request_id, result: { Ok: reply } }); } export async function fakeImage( @@ -74,5 +75,5 @@ export async function fakeImage( } export function notification(revision: bigint): string { - return json({ version: 4, event: { event: "image_changed", data: { revision } } }); + return json({ version: 5, event: { event: "image_changed", data: { revision } } }); } diff --git a/vscode/test/protocol.test.ts b/vscode/test/protocol.test.ts index 74cdba2..7d81cc6 100644 --- a/vscode/test/protocol.test.ts +++ b/vscode/test/protocol.test.ts @@ -15,12 +15,12 @@ import { actor, json, REF } from "./fixtures"; const command: Command = { operation: "list_actors" }; const envelope = (reply: unknown) => - json({ version: 4, request_id: "test", result: { Ok: reply } }); + json({ version: 5, request_id: "test", result: { Ok: reply } }); test("notification envelopes preserve u64 revisions and reject malformed events", () => { const notification = (data: unknown) => json({ - version: 4, + version: 5, event: { event: "image_changed", data, @@ -52,7 +52,7 @@ test("notification envelopes preserve u64 revisions and reject malformed events" ); }); -test("v4 snapshots preserve all value variants, i64 limits, u64 telemetry, and mailbox order", () => { +test("v5 snapshots preserve all value variants, i64 limits, u64 telemetry, and mailbox order", () => { const snapshot = actor(); const reply = decodeResponse(envelope({ kind: "actors", data: [snapshot] }), "test", command); assert.deepEqual(reply, { kind: "actors", data: [snapshot] }); @@ -78,7 +78,7 @@ test("version, correlation, result, reply kind, and actor identity are checked", assert.throws(() => decodeResponse( json({ - version: 4, + version: 5, request_id: "test", result: { Ok: { kind: "actors", data: [] } }, ...change, @@ -104,7 +104,7 @@ test("server errors are explicit, typed, and not treated as empty actor lists", () => decodeResponse( json({ - version: 4, + version: 5, request_id: "test", result: { Err: { code: "not_found", message: "No actor" } }, }), diff --git a/vscode/test/values.test.ts b/vscode/test/values.test.ts index 471c51c..32ac53a 100644 --- a/vscode/test/values.test.ts +++ b/vscode/test/values.test.ts @@ -1,8 +1,22 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { parseValue } from "../src/values"; +import { parseValue, formatValue } from "../src/values"; import { REF } from "./fixtures"; +test("editing JSON round-trips exact integers, references, null messages and escaped records", () => { + for (const source of [ + "null", + "-9223372036854775808", + "9223372036854775807", + `{"reserved":[null],"queued":[{"$actor":"${REF}"},{"$record":{"$actor":"literal"}}]}`, + '{"$record":{"$record":{"value":true}}}', + '{"__proto__":null}', + ]) { + const value = parseValue(source); + assert.deepEqual(parseValue(formatValue(value)), value); + } +}); + test("natural JSON preserves i64s, string whitespace, markers, and all value variants", () => { assert.deepEqual( parseValue(