diff --git a/AGENTS.md b/AGENTS.md index f742f78..7a0a8ea 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -68,6 +68,7 @@ proposed capabilities as plans, not existing behavior. - `crates/runa-runtime/`: single-worker scheduling and in-memory transitions. - `crates/runa-image-api/`: trusted in-process services for operating on an image. - `crates/runa-protocol/`: versioned transport-neutral request, response, and event DTOs. +- `crates/runa-server/`: transport-neutral dispatch from protocol requests to image API services. - `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. diff --git a/CHANGELOG.md b/CHANGELOG.md index b2ca9e5..e13e4b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,8 @@ All notable changes to this project will be documented in this file. operations, now used by the local CLI. - Versioned, transport-neutral `runa-protocol` request, response, and event DTOs with serde support and request validation. +- Transport-neutral `runa-server` dispatcher routing protocol commands through + the image API with typed replies and error codes. - 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 77b8776..ab2b760 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -457,6 +457,16 @@ dependencies = [ "rune", ] +[[package]] +name = "runa-server" +version = "0.0.0" +dependencies = [ + "runa-actor", + "runa-code", + "runa-image-api", + "runa-protocol", +] + [[package]] name = "rune" version = "0.14.2" diff --git a/Cargo.toml b/Cargo.toml index 709b7a1..d97aecc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,4 +9,5 @@ members = [ "crates/runa-protocol", "crates/runa-rune", "crates/runa-runtime", + "crates/runa-server", ] diff --git a/README.md b/README.md index d61fdb3..ec03710 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,8 @@ implemented yet. The code library now supports an in-memory versioned behavior registry, and the CLI exposes publishing, history, rollback, safe-boundary installation, pause/resume, and cancellation through a trusted in-process image API. A versioned transport-neutral protocol contract is now defined, but no -network server or authentication is implemented. +network transport or authentication is implemented. A transport-neutral +dispatcher routes protocol requests through the same image API. The prototype uses worker-local invocation stacks and cooperative instruction slices, and actor snapshots report basic active-execution progress. Native @@ -108,7 +109,8 @@ test task runs `cargo test --workspace`. │ ├── runa-rune/ # Phase 0 Rune experiments │ ├── runa-runtime/ # Single-worker actor prototype │ ├── runa-image-api/ # Trusted in-process image operations -│ └── runa-protocol/ # Versioned transport-neutral wire DTOs +│ ├── runa-protocol/ # Versioned transport-neutral wire DTOs +│ └── runa-server/ # Transport-neutral protocol dispatcher ├── docs/ │ ├── architecture.md │ ├── protocol.md diff --git a/crates/runa-server/Cargo.toml b/crates/runa-server/Cargo.toml new file mode 100644 index 0000000..5c357f5 --- /dev/null +++ b/crates/runa-server/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "runa-server" +version = "0.0.0" +edition = "2024" + +[dependencies] +runa-actor = { path = "../runa-actor" } +runa-code = { path = "../runa-code" } +runa-image-api = { path = "../runa-image-api" } +runa-protocol = { path = "../runa-protocol" } diff --git a/crates/runa-server/src/lib.rs b/crates/runa-server/src/lib.rs new file mode 100644 index 0000000..71d7f41 --- /dev/null +++ b/crates/runa-server/src/lib.rs @@ -0,0 +1,431 @@ +//! Transport-neutral dispatch from protocol requests to image API operations. + +use std::num::NonZeroUsize; + +use runa_actor::{ActorId, Event, Message, Status}; +use runa_image_api::ImageApi; +use runa_protocol::{ + ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode, + EventDto, ExecutionSnapshotDto, MessageDto, PROTOCOL_VERSION, ProtocolError, Reply, Request, + Response, +}; + +pub struct Server { + image: ImageApi, +} + +impl Default for Server { + fn default() -> Self { + Self::new() + } +} + +impl Server { + pub fn new() -> Self { + Self { + image: ImageApi::new(), + } + } + + pub fn with_image_api(image: ImageApi) -> Self { + Self { image } + } + + pub fn dispatch(&mut self, request: Request) -> Response { + let request_id = request.request_id.clone(); + let result = request + .validate() + .and_then(|()| self.dispatch_command(request.command)); + + Response { + version: PROTOCOL_VERSION, + request_id, + result, + } + } + + fn dispatch_command(&mut self, command: Command) -> Result { + match command { + Command::PublishBehavior { + name, + expected_revision, + source, + } => self + .image + .publish_behavior(&name, expected_revision, &source) + .map(behavior_dto) + .map(Reply::Behavior) + .map_err(api_error), + Command::RollbackBehavior { + name, + expected_revision, + target_revision, + } => self + .image + .rollback_behavior(&name, expected_revision, target_revision) + .map(behavior_dto) + .map(Reply::Behavior) + .map_err(api_error), + Command::GetBehavior { name } => self + .image + .behavior(&name) + .map(behavior_dto) + .map(Reply::Behavior) + .ok_or_else(|| ProtocolError::new(ErrorCode::NotFound, "unknown behavior")), + Command::GetBehaviors => Ok(Reply::Behaviors( + self.image + .behaviors() + .into_iter() + .map(behavior_dto) + .collect(), + )), + Command::GetBehaviorSource { name, revision } => self + .image + .behavior_source(&name, revision) + .map(|source| Reply::BehaviorSource { + source: source.to_owned(), + }) + .map_err(api_error), + Command::Spawn { + behavior_name, + state, + mailbox_capacity, + } => { + let capacity = NonZeroUsize::new(mailbox_capacity) + .ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?; + self.image + .spawn(&behavior_name, state, capacity) + .map(|actor_id| Reply::ActorSpawned { + actor_id: actor_id.0, + }) + .map_err(api_error) + } + Command::Send { actor_id, message } => self + .image + .send(ActorId(actor_id), message_from_dto(message)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::Tick { + ticks, + instruction_budget, + } => { + let budget = NonZeroUsize::new(instruction_budget) + .ok_or_else(|| invalid_request("instruction_budget must be positive"))?; + self.image.tick(ticks, budget); + Ok(Reply::TickCompleted { ticks }) + } + Command::InspectActor { actor_id } => self + .image + .inspect(ActorId(actor_id)) + .map(|snapshot| actor_snapshot_dto(&snapshot)) + .map(Reply::Actor) + .map_err(api_error), + Command::ListActors => Ok(Reply::Actors( + self.image.actors().iter().map(actor_snapshot_dto).collect(), + )), + Command::GetEvents => Ok(Reply::Events( + self.image.events().iter().map(event_dto).collect(), + )), + Command::StopActor { actor_id } => self + .image + .stop(ActorId(actor_id)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::RestartActor { actor_id } => self + .image + .restart(ActorId(actor_id)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::PauseActor { actor_id } => self + .image + .pause(ActorId(actor_id)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::ResumeActor { actor_id } => self + .image + .resume(ActorId(actor_id)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::CancelActor { actor_id } => self + .image + .cancel(ActorId(actor_id)) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::InstallBehavior { + actor_id, + behavior_name, + } => self + .image + .install(ActorId(actor_id), &behavior_name) + .map(|implementation_id| Reply::BehaviorInstalled { + implementation_id: implementation_id.to_string(), + }) + .map_err(api_error), + } + } +} + +fn behavior_dto(behavior: runa_code::BehaviorSnapshot) -> BehaviorDto { + BehaviorDto { + name: behavior.name, + revision: behavior.revision, + current_implementation_id: behavior.current.to_string(), + history: behavior + .history + .into_iter() + .map(|version| BehaviorVersionDto { + revision: version.revision, + implementation_id: version.implementation_id.to_string(), + }) + .collect(), + } +} + +fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto { + ActorSnapshotDto { + actor_id: snapshot.id.0, + mailbox_id: snapshot.mailbox.0, + implementation_id: snapshot.implementation.to_string(), + incarnation: snapshot.incarnation, + state: snapshot.state, + status: match &snapshot.status { + Status::Idle => ActorStatusDto::Idle, + Status::Running => ActorStatusDto::Running, + Status::Awaiting => ActorStatusDto::Awaiting, + Status::Paused => ActorStatusDto::Paused, + Status::Stopped => ActorStatusDto::Stopped, + Status::Failed(message) => ActorStatusDto::Failed(message.clone()), + }, + execution: snapshot.execution.map(|execution| ExecutionSnapshotDto { + vm_steps: execution.vm_steps, + instruction_pointer: execution.instruction_pointer, + call_depth: execution.call_depth, + }), + queued: snapshot.queued.iter().copied().map(message_dto).collect(), + reserved: snapshot.reserved.map(message_dto), + } +} + +fn message_dto(message: Message) -> MessageDto { + MessageDto { + kind: message.kind, + value: message.value, + } +} + +fn message_from_dto(message: MessageDto) -> Message { + Message { + kind: message.kind, + value: message.value, + } +} + +fn event_dto(event: &Event) -> EventDto { + match event { + Event::Spawned(actor_id) => EventDto::Spawned { + actor_id: actor_id.0, + }, + Event::Committed(actor_id) => EventDto::Committed { + actor_id: actor_id.0, + }, + Event::Failed(actor_id, message) => EventDto::Failed { + actor_id: actor_id.0, + message: message.clone(), + }, + Event::Stopped(actor_id) => EventDto::Stopped { + actor_id: actor_id.0, + }, + Event::Restarted(actor_id) => EventDto::Restarted { + actor_id: actor_id.0, + }, + Event::Paused(actor_id) => EventDto::Paused { + actor_id: actor_id.0, + }, + Event::Resumed(actor_id) => EventDto::Resumed { + actor_id: actor_id.0, + }, + Event::InvocationCancelled(actor_id) => EventDto::InvocationCancelled { + actor_id: actor_id.0, + }, + Event::BehaviorInstalled(actor_id, implementation_id) => EventDto::BehaviorInstalled { + actor_id: actor_id.0, + implementation_id: implementation_id.to_string(), + }, + Event::StateMigrated(actor_id, previous_state, current_state) => EventDto::StateMigrated { + actor_id: actor_id.0, + previous_state: *previous_state, + current_state: *current_state, + }, + } +} + +fn api_error(error: impl std::fmt::Display) -> ProtocolError { + let message = error.to_string(); + let code = if message.starts_with("behavior revision conflict:") { + ErrorCode::Conflict + } else if matches!( + message.as_str(), + "unknown actor" + | "unknown behavior" + | "unknown behavior revision" + | "unknown target behavior revision" + | "unknown implementation" + ) { + ErrorCode::NotFound + } else if matches!( + message.as_str(), + "mailbox full" | "effect exceeds mailbox capacity" + ) { + ErrorCode::CapacityExceeded + } else if message == "invalid behavior name" { + ErrorCode::InvalidRequest + } else { + ErrorCode::RuntimeFailure + }; + ProtocolError::new(code, message) +} + +fn invalid_request(message: impl Into) -> ProtocolError { + ProtocolError::new(ErrorCode::InvalidRequest, message) +} + +#[cfg(test)] +mod tests { + use super::*; + use runa_protocol::{Command, PROTOCOL_VERSION}; + + fn request(command: Command) -> Request { + Request { + version: PROTOCOL_VERSION, + request_id: "test-request".to_owned(), + command, + } + } + + fn publish(server: &mut Server) -> BehaviorDto { + let response = server.dispatch(request(Command::PublishBehavior { + name: "counter".to_owned(), + expected_revision: None, + source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned(), + })); + match response.result.unwrap() { + Reply::Behavior(behavior) => behavior, + other => panic!("unexpected reply: {other:?}"), + } + } + + #[test] + fn dispatches_behavior_actor_and_event_operations() { + let mut server = Server::new(); + let behavior = publish(&mut server); + assert_eq!(behavior.revision, 1); + assert_eq!(behavior.history.len(), 1); + + let spawned = server.dispatch(request(Command::Spawn { + behavior_name: "counter".to_owned(), + state: 5, + mailbox_capacity: 2, + })); + let actor_id = match spawned.result.unwrap() { + Reply::ActorSpawned { actor_id } => actor_id, + other => panic!("unexpected reply: {other:?}"), + }; + assert_eq!(actor_id, 1); + + let sent = server.dispatch(request(Command::Send { + actor_id, + message: MessageDto { kind: 0, value: 7 }, + })); + assert_eq!(sent.result, Ok(Reply::Acknowledged)); + server.dispatch(request(Command::Tick { + ticks: 100, + instruction_budget: 10, + })); + + let inspected = server.dispatch(request(Command::InspectActor { actor_id })); + match inspected.result.unwrap() { + Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12), + other => panic!("unexpected reply: {other:?}"), + } + + let events = server.dispatch(request(Command::GetEvents)); + match events.result.unwrap() { + Reply::Events(events) => assert!(events.iter().any( + |event| matches!(event, EventDto::Committed { actor_id: id } if *id == actor_id) + )), + other => panic!("unexpected reply: {other:?}"), + } + assert_eq!( + server.dispatch(request(Command::GetEvents)).result, + Ok(Reply::Events(Vec::new())) + ); + } + + #[test] + fn dispatch_preserves_request_id_and_maps_validation_and_domain_errors() { + let mut server = Server::new(); + let mut invalid = request(Command::GetEvents); + invalid.version += 1; + let response = server.dispatch(invalid); + assert_eq!(response.request_id, "test-request"); + assert_eq!( + response.result.unwrap_err().code, + ErrorCode::UnsupportedVersion + ); + + let missing = server.dispatch(request(Command::InspectActor { actor_id: 42 })); + assert_eq!(missing.result.unwrap_err().code, ErrorCode::NotFound); + + publish(&mut server); + let conflict = server.dispatch(request(Command::PublishBehavior { + name: "counter".to_owned(), + expected_revision: Some(99), + source: "pub fn handle(state, kind, value) { (state, []) }".to_owned(), + })); + assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict); + } + + #[test] + fn dispatch_supports_behavior_history_source_and_safe_install() { + let mut server = Server::new(); + let first = publish(&mut server); + let updated = server.dispatch(request(Command::PublishBehavior { + name: "counter".to_owned(), + expected_revision: Some(first.revision), + source: "pub fn handle(state, kind, value) { (state * value, []) }".to_owned(), + })); + assert!(matches!(updated.result, Ok(Reply::Behavior(_)))); + + let source = server.dispatch(request(Command::GetBehaviorSource { + name: "counter".to_owned(), + revision: 1, + })); + assert_eq!( + source.result, + Ok(Reply::BehaviorSource { + source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned() + }) + ); + + let actor_id = match server + .dispatch(request(Command::Spawn { + behavior_name: "counter".to_owned(), + state: 3, + mailbox_capacity: 1, + })) + .result + .unwrap() + { + Reply::ActorSpawned { actor_id } => actor_id, + other => panic!("unexpected reply: {other:?}"), + }; + let installed = server.dispatch(request(Command::InstallBehavior { + actor_id, + behavior_name: "counter".to_owned(), + })); + assert!(matches!( + installed.result, + Ok(Reply::BehaviorInstalled { .. }) + )); + } +} diff --git a/docs/architecture.md b/docs/architecture.md index bb1edcc..a6aa41a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -310,7 +310,7 @@ 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` | WebSocket/JSON-RPC transport, sessions, authentication, subscriptions | `runa-protocol`, `runa-image-api` | +| `runa-server` | Protocol request dispatch and, in a later increment, transport, sessions, authentication, subscriptions | `runa-protocol`, `runa-image-api` | | `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 | @@ -327,7 +327,10 @@ publication, history and rollback, actor lifecycle and messaging, snapshots, ticks, and event retrieval. The CLI uses these application services rather than reaching into the code store and runtime directly. `runa-protocol` now defines versioned, serializable request, response, and event DTOs without -depending on a transport or the runtime implementation. +depending on a transport or the runtime implementation. `runa-server` now +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 transport, session model, authentication, or authorization. **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. @@ -432,9 +435,9 @@ inspection remain future work. - CLI client that can spawn, inspect, send messages, and replace behaviors. The first Phase 5 increments are a trusted in-process application API used by -the CLI and a transport-neutral versioned protocol contract. The protocol is a -Rust DTO crate with serde serialization; a network server and authentication -are not yet implemented. +the CLI, a transport-neutral versioned protocol contract, and an in-process +dispatcher from protocol commands to application services. A network server +and authentication are not yet implemented. ### Phase 6 — Graphical world diff --git a/docs/protocol.md b/docs/protocol.md index e72e98e..87dec90 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -1,9 +1,9 @@ # Runa Protocol `runa-protocol` defines versioned request, response, and event data contracts -for clients and a future server. It is independent of transport and runtime -implementation; the current crate does not dispatch requests, expose a network -endpoint, or provide authentication. +for clients and the `runa-server` dispatcher. It is independent of transport +and runtime implementation; the server currently dispatches requests +in-process, with no network endpoint or authentication. ## Version 1 @@ -38,9 +38,14 @@ portable wire values. Responses echo the request ID and carry either a typed reply or a protocol error. Events use a separate versioned envelope. Actor and behavior snapshots are wire DTOs, not serialized runtime objects; implementation identities are -represented by their stable display strings. Runtime errors and validation -errors must be mapped by a future server to the defined error codes without -claiming a more specific category than it can establish. +represented by their stable display strings. The server maps known service +failures to the defined error codes and otherwise uses `runtime_failure` +rather than claiming a more specific category. + +`get_events` drains and returns the events currently accumulated by this +single-worker image. It is not a durable subscription or replayable event log. +Any future network transport must also impose message and request-size limits +before dispatch; the in-process protocol crate does not impose byte limits. The current validation checks the protocol version, request ID, positive actor/revision identifiers where required, non-empty behavior names, mailbox diff --git a/docs/runtime.md b/docs/runtime.md index 4f38a5e..658e5f5 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -51,6 +51,8 @@ The default per-actor slice budget is 32 instructions. trusted in-process service boundary. - `runa-protocol` defines versioned transport-neutral request, response, and event DTOs; it does not dispatch requests or access runtime state. +- `runa-server` dispatches those requests through `runa-image-api` and maps + results to protocol replies; it currently has no network transport. - `runa` provides the local CLI client for that boundary. The image API is process-local and intended for trusted callers. It has no