diff --git a/AGENTS.md b/AGENTS.md index 81d2b26..4d271fa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -69,6 +69,7 @@ proposed capabilities as plans, not existing behavior. - `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 and local JSON-lines stdio transport. +- `crates/runa-client/`: local child-process client for the JSON-lines protocol. - `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. @@ -88,6 +89,7 @@ mise run format mise run lint mise run test mise run run +mise run serve mise run pre-commit ``` diff --git a/CHANGELOG.md b/CHANGELOG.md index 07d39cc..717c19a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,8 @@ All notable changes to this project will be documented in this file. the image API with typed replies and error codes. - Local JSON-lines stdio transport for the protocol dispatcher, with a 1 MiB request limit and explicit malformed-input failures. +- Reusable local child-process client for issuing correlated protocol requests + and shutting down the server process cleanly. - Execution-model decision and Phase 1 runtime documentation. - Lifecycle tests proving failed-message retry from committed state and stop/restart preservation of reserved-first FIFO order. diff --git a/Cargo.lock b/Cargo.lock index 6ee1d0d..2fd9c30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -403,6 +403,14 @@ dependencies = [ "runa-core", ] +[[package]] +name = "runa-client" +version = "0.0.0" +dependencies = [ + "runa-protocol", + "serde_json", +] + [[package]] name = "runa-code" version = "0.0.0" @@ -458,6 +466,7 @@ name = "runa-server" version = "0.0.0" dependencies = [ "runa-actor", + "runa-client", "runa-code", "runa-image-api", "runa-protocol", diff --git a/Cargo.toml b/Cargo.toml index d97aecc..2fe80b2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "3" members = [ "crates/runa", "crates/runa-actor", + "crates/runa-client", "crates/runa-code", "crates/runa-core", "crates/runa-image-api", diff --git a/README.md b/README.md index 571fd8c..78d77bd 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,11 @@ one JSON request per line on stdin; responses are written to stdout. Request lines are limited to 1 MiB. This is not a network server and is intended for a trusted parent process. +The `runa-client` crate can launch that process and exchange typed protocol +requests with request/response correlation. The VS Code extension is intended +to use an authenticated network connection to this same live image; actors +remain in the image regardless of how clients connect. + The CLI opens an interactive runtime session using the default counter behavior: ```text @@ -117,7 +122,8 @@ test task runs `cargo test --workspace`. │ ├── 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-server/ # Protocol dispatcher and local JSON-lines transport +│ └── runa-client/ # Local child-process protocol client ├── docs/ │ ├── architecture.md │ ├── protocol.md diff --git a/crates/runa-client/Cargo.toml b/crates/runa-client/Cargo.toml new file mode 100644 index 0000000..3fb5d2d --- /dev/null +++ b/crates/runa-client/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "runa-client" +version = "0.0.0" +edition = "2024" + +[dependencies] +runa-protocol = { path = "../runa-protocol" } +serde_json = "1" diff --git a/crates/runa-client/src/lib.rs b/crates/runa-client/src/lib.rs new file mode 100644 index 0000000..f652999 --- /dev/null +++ b/crates/runa-client/src/lib.rs @@ -0,0 +1,294 @@ +//! Local child-process client for the versioned JSON-lines protocol. + +use std::ffi::OsStr; +use std::fmt; +use std::io::{self, BufRead, BufReader, BufWriter, Write}; +use std::process::{Child, ChildStdin, ChildStdout, Command, ExitStatus, Stdio}; + +use runa_protocol::{Command as ProtocolCommand, PROTOCOL_VERSION, Request, Response}; + +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, +} + +#[derive(Debug)] +pub enum ClientError { + Io(io::Error), + Serialization(serde_json::Error), + RequestIdsExhausted, + RequestTooLarge, + ResponseTooLarge, + UnsupportedVersion(u16), + CorrelationMismatch { expected: String, actual: String }, +} + +impl fmt::Display for ClientError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io(error) => write!(formatter, "local protocol transport failed: {error}"), + Self::Serialization(error) => write!(formatter, "protocol JSON failed: {error}"), + 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" + ), + Self::UnsupportedVersion(version) => { + write!( + formatter, + "server returned unsupported protocol version {version}" + ) + } + Self::CorrelationMismatch { expected, actual } => write!( + formatter, + "response request ID mismatch: expected `{expected}`, got `{actual}`" + ), + } + } +} + +impl std::error::Error for ClientError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(error) => Some(error), + Self::Serialization(error) => Some(error), + _ => None, + } + } +} + +impl From for ClientError { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} + +impl From for ClientError { + fn from(error: serde_json::Error) -> Self { + Self::Serialization(error) + } +} + +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 stdout 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; + + 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 index 183f463..1266400 100644 --- a/crates/runa-server/Cargo.toml +++ b/crates/runa-server/Cargo.toml @@ -9,3 +9,6 @@ runa-code = { path = "../runa-code" } runa-image-api = { path = "../runa-image-api" } runa-protocol = { path = "../runa-protocol" } serde_json = "1" + +[dev-dependencies] +runa-client = { path = "../runa-client" } diff --git a/crates/runa-server/tests/local_client.rs b/crates/runa-server/tests/local_client.rs new file mode 100644 index 0000000..ac735ec --- /dev/null +++ b/crates/runa-server/tests/local_client.rs @@ -0,0 +1,48 @@ +use runa_client::LocalClient; +use runa_protocol::{Command, MessageDto, Reply}; + +#[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, kind, value) { (state + 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: 5, + mailbox_capacity: 2, + }) + .unwrap(); + assert_eq!(spawned.request_id, "runa-client-2"); + assert_eq!(spawned.result, Ok(Reply::ActorSpawned { actor_id: 1 })); + + client + .request(Command::Send { + actor_id: 1, + message: MessageDto { kind: 0, value: 7 }, + }) + .unwrap(); + client + .request(Command::Tick { + ticks: 100, + instruction_budget: 10, + }) + .unwrap(); + let inspected = client + .request(Command::InspectActor { actor_id: 1 }) + .unwrap(); + assert!(matches!( + inspected.result, + Ok(Reply::Actor(actor)) if actor.state == 12 + )); + assert!(client.shutdown().unwrap().success()); +} diff --git a/docs/architecture.md b/docs/architecture.md index 993489a..41c945c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -5,6 +5,7 @@ **Initial embedded language:** Rune **Inspirations:** Smalltalk/Squeak/Pharo, Morphic, Erlang/OTP **Primary interfaces:** Self-contained graphical world and VS Code extension +connected to the same live image through an authenticated network API ## 1. Vision @@ -62,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 uses a versioned external transport. Neither interface should have privileged ad hoc paths that violate runtime invariants. +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 network boundary must authenticate and authorize clients; it does not create a separate actor runtime or image. ## 4. Fundamental Actor Model @@ -194,7 +195,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 both the graphical tools and the VS Code extension. It must also be usable by CLI tools, tests, automation, and coding agents. +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. Suggested operation families (provisional names): @@ -232,7 +233,7 @@ Normal actors should not automatically receive development privileges. Arbitrary ## 9. VS Code Extension -VS Code is an external client of the same live image, not a separate runtime. Begin with JSON-RPC over local WebSocket or an equivalently straightforward versioned transport. +VS Code is an external client of the same live image, not a separate runtime. It connects through an authenticated versioned network transport, initially planned as JSON-RPC over WebSocket. 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. Network security, connection lifecycle, and event subscription semantics must be designed before exposing the listener. Required extension UX: @@ -310,7 +311,8 @@ 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 and local JSON-lines stdio transport; later network transports, sessions, authentication, subscriptions | `runa-protocol`, `runa-image-api` | +| `runa-server` | Protocol dispatch and local JSON-lines stdio transport; authenticated network transport for VS Code clients | `runa-protocol`, `runa-image-api` | +| `runa-client` | Local child-process client for the JSON-lines stdio protocol | `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 | @@ -332,7 +334,11 @@ dispatches those commands through the image API and maps service results into protocol replies and typed errors. It remains an in-process dispatcher; there is no network listener, session model, authentication, or authorization. The dispatcher is exposed through local child-process JSON-lines stdio with a 1 -MiB request-line limit. +MiB request-line limit. In-process consumers call `runa-image-api`, and +external local tools can use `runa-client` to launch the stdio server. Actors +do not depend on network transport to coexist in one live image; the network +boundary exists for external clients such as VS Code. Authenticated network +transport is a planned integration requirement, not a runtime prerequisite. **Important:** `runa-core` must not contain schedulers, mailbox implementations, persistence engines, Rune VM types, service traits used by only one crate, generic helper collections, or GUI abstractions. Prefer a type defined in its owning crate until two independent consumers demonstrably need a stable shared representation. @@ -371,6 +377,7 @@ runa/ │ ├── runa-image-api/ │ ├── runa-protocol/ │ ├── runa-server/ +│ ├── runa-client/ │ ├── runa-graphics/ │ ├── runa-world/ │ └── runa/ @@ -439,8 +446,10 @@ 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 uses the -same protocol commands and dispatcher in-process. A network server, sessions, -authentication, and authorization are not yet implemented. +same protocol commands and dispatcher in-process. `runa-client` launches the +stdio server for external local tools. The next server increment for the VS +Code integration is an authenticated network transport over the same API and +live image; it is not needed for actors to share that image. ### Phase 6 — Graphical world diff --git a/docs/protocol.md b/docs/protocol.md index ab9e3fa..32a7234 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -49,8 +49,11 @@ single-worker image. It is not a durable subscription or replayable event log. The stdio transport limits each request line to 1 MiB before parsing. Malformed JSON or an oversized line terminates the stream with an error on stderr; no protocol response is emitted when a request cannot be decoded. EOF cleanly -terminates the process. Any future network transport must impose its own -message and request-size limits before dispatch. +terminates the process. The `runa-client` crate launches this server locally +and validates response protocol versions and request IDs. The VS Code +extension is planned to use a separate authenticated network transport to +the same image; any such transport must impose its own message and +request-size limits before dispatch. The current validation checks the protocol version, request ID, positive actor/revision identifiers where required, non-empty behavior names, mailbox