runa/crates/runa-protocol/src/lib.rs

//! Versioned, transport-neutral wire contracts for Runa clients and servers.

use serde::{Deserialize, Serialize};

pub const PROTOCOL_VERSION: u16 = 1;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Request {
    pub version: u16,
    pub request_id: String,
    pub command: Command,
}

impl Request {
    pub fn validate(&self) -> Result<(), ProtocolError> {
        if self.version != PROTOCOL_VERSION {
            return Err(ProtocolError::new(
                ErrorCode::UnsupportedVersion,
                format!("unsupported protocol version {}", self.version),
            ));
        }
        if self.request_id.trim().is_empty() {
            return Err(ProtocolError::new(
                ErrorCode::InvalidRequest,
                "request_id must not be empty",
            ));
        }

        match &self.command {
            Command::PublishBehavior {
                name,
                expected_revision,
                ..
            } => {
                validate_name(name)?;
                validate_optional_revision(*expected_revision)?;
            }
            Command::RollbackBehavior {
                name,
                expected_revision,
                target_revision,
            } => {
                validate_name(name)?;
                validate_revision(*expected_revision)?;
                validate_revision(*target_revision)?;
            }
            Command::GetBehavior { name } => validate_name(name)?,
            Command::GetBehaviorSource { name, revision } => {
                validate_name(name)?;
                validate_revision(*revision)?;
            }
            Command::Spawn {
                behavior_name,
                mailbox_capacity,
                ..
            } => {
                validate_name(behavior_name)?;
                if *mailbox_capacity == 0 {
                    return Err(invalid_request("mailbox_capacity must be positive"));
                }
            }
            Command::Send { actor_id, .. }
            | Command::InspectActor { actor_id }
            | Command::StopActor { actor_id }
            | Command::RestartActor { actor_id }
            | Command::PauseActor { actor_id }
            | Command::ResumeActor { actor_id }
            | Command::CancelActor { actor_id } => validate_actor_id(*actor_id)?,
            Command::InstallBehavior {
                actor_id,
                behavior_name,
            } => {
                validate_actor_id(*actor_id)?;
                validate_name(behavior_name)?;
            }
            Command::Tick {
                instruction_budget, ..
            } if *instruction_budget == 0 => {
                return Err(invalid_request("instruction_budget must be positive"));
            }
            Command::GetBehaviors
            | Command::ListActors
            | Command::GetEvents
            | Command::Tick { .. } => {}
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "operation", content = "parameters", rename_all = "snake_case")]
pub enum Command {
    PublishBehavior {
        name: String,
        expected_revision: Option<u64>,
        source: String,
    },
    RollbackBehavior {
        name: String,
        expected_revision: u64,
        target_revision: u64,
    },
    GetBehavior {
        name: String,
    },
    GetBehaviors,
    GetBehaviorSource {
        name: String,
        revision: u64,
    },
    Spawn {
        behavior_name: String,
        state: i64,
        mailbox_capacity: usize,
    },
    Send {
        actor_id: u64,
        message: MessageDto,
    },
    Tick {
        ticks: usize,
        instruction_budget: usize,
    },
    InspectActor {
        actor_id: u64,
    },
    ListActors,
    GetEvents,
    StopActor {
        actor_id: u64,
    },
    RestartActor {
        actor_id: u64,
    },
    PauseActor {
        actor_id: u64,
    },
    ResumeActor {
        actor_id: u64,
    },
    CancelActor {
        actor_id: u64,
    },
    InstallBehavior {
        actor_id: u64,
        behavior_name: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Response {
    pub version: u16,
    pub request_id: String,
    pub result: Result<Reply, ProtocolError>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum Reply {
    Acknowledged,
    Behavior(BehaviorDto),
    Behaviors(Vec<BehaviorDto>),
    BehaviorSource { source: String },
    ActorSpawned { actor_id: u64 },
    Actor(ActorSnapshotDto),
    Actors(Vec<ActorSnapshotDto>),
    Events(Vec<EventDto>),
    TickCompleted { ticks: usize },
    BehaviorInstalled { implementation_id: String },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ProtocolEvent {
    pub version: u16,
    pub event: EventDto,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "event", content = "data", rename_all = "snake_case")]
pub enum EventDto {
    Spawned {
        actor_id: u64,
    },
    Committed {
        actor_id: u64,
    },
    Failed {
        actor_id: u64,
        message: String,
    },
    Stopped {
        actor_id: u64,
    },
    Restarted {
        actor_id: u64,
    },
    Paused {
        actor_id: u64,
    },
    Resumed {
        actor_id: u64,
    },
    InvocationCancelled {
        actor_id: u64,
    },
    BehaviorInstalled {
        actor_id: u64,
        implementation_id: String,
    },
    StateMigrated {
        actor_id: u64,
        previous_state: i64,
        current_state: i64,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BehaviorDto {
    pub name: String,
    pub revision: u64,
    pub current_implementation_id: String,
    pub history: Vec<BehaviorVersionDto>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BehaviorVersionDto {
    pub revision: u64,
    pub implementation_id: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MessageDto {
    pub kind: i64,
    pub value: i64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ActorSnapshotDto {
    pub actor_id: u64,
    pub mailbox_id: u64,
    pub implementation_id: String,
    pub incarnation: u64,
    pub state: i64,
    pub status: ActorStatusDto,
    pub execution: Option<ExecutionSnapshotDto>,
    pub queued: Vec<MessageDto>,
    pub reserved: Option<MessageDto>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "status", content = "message", rename_all = "snake_case")]
pub enum ActorStatusDto {
    Idle,
    Running,
    Awaiting,
    Paused,
    Stopped,
    Failed(String),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ExecutionSnapshotDto {
    pub vm_steps: u64,
    pub instruction_pointer: usize,
    pub call_depth: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ErrorCode {
    InvalidRequest,
    UnsupportedVersion,
    NotFound,
    Conflict,
    CapacityExceeded,
    RuntimeFailure,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ProtocolError {
    pub code: ErrorCode,
    pub message: String,
}

impl ProtocolError {
    pub fn new(code: ErrorCode, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
        }
    }
}

fn validate_name(name: &str) -> Result<(), ProtocolError> {
    if name.trim().is_empty() {
        return Err(invalid_request("behavior name must not be empty"));
    }
    Ok(())
}

fn validate_actor_id(actor_id: u64) -> Result<(), ProtocolError> {
    if actor_id == 0 {
        return Err(invalid_request("actor_id must be positive"));
    }
    Ok(())
}

fn validate_revision(revision: u64) -> Result<(), ProtocolError> {
    if revision == 0 {
        return Err(invalid_request("revision must be positive"));
    }
    Ok(())
}

fn validate_optional_revision(revision: Option<u64>) -> Result<(), ProtocolError> {
    if let Some(revision) = revision {
        validate_revision(revision)?;
    }
    Ok(())
}

fn invalid_request(message: impl Into<String>) -> ProtocolError {
    ProtocolError::new(ErrorCode::InvalidRequest, message)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn request(command: Command) -> Request {
        Request {
            version: PROTOCOL_VERSION,
            request_id: "req-1".to_owned(),
            command,
        }
    }

    #[test]
    fn request_uses_a_stable_tagged_wire_shape() {
        let value = serde_json::to_value(request(Command::Send {
            actor_id: 7,
            message: MessageDto { kind: 2, value: 9 },
        }))
        .unwrap();
        assert_eq!(
            value,
            serde_json::json!({
                "version": PROTOCOL_VERSION,
                "request_id": "req-1",
                "command": {
                    "operation": "send",
                    "parameters": {
                        "actor_id": 7,
                        "message": { "kind": 2, "value": 9 }
                    }
                }
            })
        );
    }

    #[test]
    fn requests_validate_versions_and_required_positive_values() {
        let mut request = request(Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: 0,
            mailbox_capacity: 1,
        });
        assert_eq!(request.validate(), Ok(()));

        request.version += 1;
        assert_eq!(
            request.validate().unwrap_err().code,
            ErrorCode::UnsupportedVersion
        );

        request.version = PROTOCOL_VERSION;
        request.command = Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: 0,
            mailbox_capacity: 0,
        };
        assert_eq!(
            request.validate().unwrap_err().code,
            ErrorCode::InvalidRequest
        );
    }

    #[test]
    fn requests_reject_empty_correlation_ids_and_zero_actor_ids() {
        let mut request = request(Command::GetEvents);
        request.request_id = "  ".to_owned();
        assert_eq!(
            request.validate().unwrap_err().code,
            ErrorCode::InvalidRequest
        );

        request.request_id = "req-2".to_owned();
        request.command = Command::StopActor { actor_id: 0 };
        assert_eq!(
            request.validate().unwrap_err().code,
            ErrorCode::InvalidRequest
        );
    }

    #[test]
    fn response_round_trips_with_typed_error_or_reply() {
        let response = Response {
            version: PROTOCOL_VERSION,
            request_id: "req-1".to_owned(),
            result: Err(ProtocolError::new(ErrorCode::NotFound, "unknown actor")),
        };
        let encoded = serde_json::to_string(&response).unwrap();
        let decoded: Response = serde_json::from_str(&encoded).unwrap();
        assert_eq!(decoded, response);
    }

    #[test]
    fn event_envelopes_are_versioned_and_serializable() {
        let event = ProtocolEvent {
            version: PROTOCOL_VERSION,
            event: EventDto::StateMigrated {
                actor_id: 3,
                previous_state: 4,
                current_state: 40,
            },
        };
        let encoded = serde_json::to_string(&event).unwrap();
        let decoded: ProtocolEvent = serde_json::from_str(&encoded).unwrap();
        assert_eq!(decoded, event);
    }
}