ruchy/

actors.rs

//! MCP-compatible Actor system implementation
//!
//! Based on SPECIFICATION.md section 7: MCP Message-Passing Architecture

use anyhow::Result;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing;

/// Core Actor trait compatible with MCP message passing
#[async_trait::async_trait]
pub trait Actor: Send + Sync + 'static + Sized {
    type Message: McpSerializable + Send + 'static;
    type Response: McpSerializable + Send + 'static;

    async fn receive(&mut self, msg: Self::Message) -> Option<Self::Response>;

    /// Spawn the actor and return a handle to communicate with it
    fn spawn(mut self) -> ActorHandle<Self::Message, Self::Response> {
        let (tx, mut rx) = mpsc::channel::<(Self::Message, mpsc::Sender<Self::Response>)>(100);

        tokio::spawn(async move {
            while let Some((msg, reply_tx)) = rx.recv().await {
                let response = self.receive(msg).await;
                if let Some(resp) = response {
                    let _ = reply_tx.send(resp).await;
                }
            }
        });

        ActorHandle { tx }
    }
}

/// Trait for MCP serializable messages
pub trait McpSerializable: Serialize + for<'de> Deserialize<'de> + fmt::Debug + Clone {}

// Blanket implementation for types that satisfy the bounds
impl<T> McpSerializable for T where T: Serialize + for<'de> Deserialize<'de> + fmt::Debug + Clone {}

/// Handle for communicating with an actor
pub struct ActorHandle<M, R> {
    tx: mpsc::Sender<(M, mpsc::Sender<R>)>,
}

impl<M, R> ActorHandle<M, R>
where
    M: McpSerializable + Send + 'static,
    R: McpSerializable + Send + 'static,
{
    /// Send a message to the actor without waiting for response
    ///
    /// # Errors
    ///
    /// Returns an error if the actor has stopped and can no longer receive messages
    pub async fn send(&self, msg: M) -> Result<()> {
        let (reply_tx, _) = mpsc::channel::<R>(1);
        self.tx
            .send((msg, reply_tx))
            .await
            .map_err(|_| anyhow::anyhow!("Actor has stopped"))?;
        Ok(())
    }

    /// Send a message and wait for response
    ///
    /// # Errors
    ///
    /// Returns an error if the actor has stopped or does not respond
    pub async fn ask(&self, msg: M) -> Result<R> {
        let (reply_tx, mut reply_rx) = mpsc::channel::<R>(1);

        self.tx
            .send((msg, reply_tx))
            .await
            .map_err(|_| anyhow::anyhow!("Actor has stopped"))?;

        reply_rx
            .recv()
            .await
            .ok_or_else(|| anyhow::anyhow!("No response received"))
    }

    /// Check if the actor is still alive
    pub fn is_alive(&self) -> bool {
        !self.tx.is_closed()
    }
}

/// MCP protocol message structure
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpMessage {
    pub jsonrpc: String,
    pub method: String,
    pub params: serde_json::Value,
    pub id: Option<String>,
}

/// MCP protocol response structure
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpResponse {
    pub jsonrpc: String,
    pub result: Option<serde_json::Value>,
    pub error: Option<McpError>,
    pub id: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpError {
    pub code: i32,
    pub message: String,
    pub data: Option<serde_json::Value>,
}

/// MCP-compatible actor for handling protocol messages
pub struct McpActor {
    pub tools: Vec<String>,
}

impl McpActor {
    /// Create a new MCP actor with default tools
    ///
    /// # Examples
    ///
    /// ```
    /// use ruchy::McpActor;
    ///
    /// let actor = McpActor::new();
    /// // Actor starts with three default tools
    /// assert_eq!(actor.tools.len(), 3);
    /// ```
    pub fn new() -> Self {
        Self {
            tools: vec![
                "transpile".to_string(),
                "parse".to_string(),
                "analyze".to_string(),
            ],
        }
    }

    fn list_tools(&self) -> McpResponse {
        McpResponse {
            jsonrpc: "2.0".to_string(),
            result: Some(serde_json::json!({
                "tools": self.tools.iter().map(|name| {
                    serde_json::json!({
                        "name": name,
                        "description": format!("Ruchy {name} tool")
                    })
                }).collect::<Vec<_>>()
            })),
            error: None,
            id: None,
        }
    }

    fn call_tool(params: &serde_json::Value) -> Option<McpResponse> {
        // Extract tool name and arguments from params
        let tool_name = params.get("name")?.as_str()?;

        let result = match tool_name {
            "transpile" => {
                serde_json::json!({
                    "content": [
                        {
                            "type": "text",
                            "text": "Transpilation functionality placeholder"
                        }
                    ]
                })
            }
            "parse" => {
                serde_json::json!({
                    "content": [
                        {
                            "type": "text",
                            "text": "Parsing functionality placeholder"
                        }
                    ]
                })
            }
            "analyze" => {
                serde_json::json!({
                    "content": [
                        {
                            "type": "text",
                            "text": "Analysis functionality placeholder"
                        }
                    ]
                })
            }
            _ => {
                return Some(McpResponse {
                    jsonrpc: "2.0".to_string(),
                    result: None,
                    error: Some(McpError {
                        code: -32601,
                        message: format!("Unknown tool: {tool_name}"),
                        data: None,
                    }),
                    id: None,
                });
            }
        };

        Some(McpResponse {
            jsonrpc: "2.0".to_string(),
            result: Some(result),
            error: None,
            id: None,
        })
    }
}

impl Default for McpActor {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait::async_trait]
impl Actor for McpActor {
    type Message = McpMessage;
    type Response = McpResponse;

    async fn receive(&mut self, msg: McpMessage) -> Option<McpResponse> {
        match msg.method.as_str() {
            "tools/list" => Some(self.list_tools()),
            "tools/call" => Self::call_tool(&msg.params),
            _ => Some(McpResponse {
                jsonrpc: "2.0".to_string(),
                result: None,
                error: Some(McpError {
                    code: -32601,
                    message: format!("Unknown method: {method}", method = msg.method),
                    data: None,
                }),
                id: msg.id,
            }),
        }
    }
}

/// Supervision strategies for actor fault tolerance
#[derive(Debug, Clone)]
pub enum SupervisionStrategy {
    /// Restart only the failed child
    OneForOne,
    /// Restart all children when one fails
    OneForAll,
    /// Restart the failed child and all children started after it
    RestForOne,
}

/// Supervisor for managing actor lifecycles
pub struct Supervisor<A: Actor> {
    children: Vec<ActorHandle<A::Message, A::Response>>,
    strategy: SupervisionStrategy,
}

impl<A: Actor> Supervisor<A> {
    pub fn new(strategy: SupervisionStrategy) -> Self {
        Self {
            children: Vec::new(),
            strategy,
        }
    }

    pub fn supervise(&mut self, actor: A) {
        let handle = actor.spawn();
        self.children.push(handle);
    }

    pub async fn monitor(&mut self) {
        // Monitoring implementation would go here
        // For now, just check if actors are alive periodically
        loop {
            for (i, child) in self.children.iter().enumerate() {
                if !child.is_alive() {
                    match self.strategy {
                        SupervisionStrategy::OneForOne => {
                            tracing::warn!("Child actor {i} died, would restart in production");
                        }
                        SupervisionStrategy::OneForAll => {
                            tracing::warn!("Child actor {i} died, would restart all in production");
                        }
                        SupervisionStrategy::RestForOne => {
                            tracing::warn!(
                                "Child actor {i} died, would restart from {i} in production"
                            );
                        }
                    }
                }
            }

            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    }
}

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

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct TestMessage {
        content: String,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct TestResponse {
        echo: String,
    }

    struct EchoActor;

    #[async_trait::async_trait]
    impl Actor for EchoActor {
        type Message = TestMessage;
        type Response = TestResponse;

        async fn receive(&mut self, msg: TestMessage) -> Option<TestResponse> {
            Some(TestResponse {
                echo: format!("Echo: {content}", content = msg.content),
            })
        }
    }

    #[tokio::test]
    async fn test_actor_spawn_and_communication() -> Result<(), Box<dyn std::error::Error>> {
        let actor = EchoActor;
        let handle = actor.spawn();

        let msg = TestMessage {
            content: "Hello, Actor!".to_string(),
        };

        let response = handle.ask(msg).await?;
        assert_eq!(response.echo, "Echo: Hello, Actor!");
        Ok(())
    }

    #[tokio::test]
    async fn test_mcp_actor_list_tools() -> Result<(), Box<dyn std::error::Error>> {
        let actor = McpActor::new();
        let handle = actor.spawn();

        let msg = McpMessage {
            jsonrpc: "2.0".to_string(),
            method: "tools/list".to_string(),
            params: serde_json::Value::Null,
            id: Some("test".to_string()),
        };

        let response = handle.ask(msg).await?;
        assert!(response.result.is_some());
        assert!(response.error.is_none());
        Ok(())
    }

    #[tokio::test]
    async fn test_mcp_actor_call_tool() -> Result<(), Box<dyn std::error::Error>> {
        let actor = McpActor::new();
        let handle = actor.spawn();

        let msg = McpMessage {
            jsonrpc: "2.0".to_string(),
            method: "tools/call".to_string(),
            params: serde_json::json!({
                "name": "transpile",
                "arguments": {}
            }),
            id: Some("test".to_string()),
        };

        let response = handle.ask(msg).await?;
        assert!(response.result.is_some());
        assert!(response.error.is_none());
        Ok(())
    }

    #[test]
    fn test_supervision_strategy_creation() {
        let supervisor: Supervisor<EchoActor> = Supervisor::new(SupervisionStrategy::OneForOne);
        assert!(matches!(
            supervisor.strategy,
            SupervisionStrategy::OneForOne
        ));
        assert_eq!(supervisor.children.len(), 0);
    }

    #[test]
    fn test_mcp_message_serialization() -> Result<(), Box<dyn std::error::Error>> {
        let msg = McpMessage {
            jsonrpc: "2.0".to_string(),
            method: "test".to_string(),
            params: serde_json::json!({"key": "value"}),
            id: Some("123".to_string()),
        };

        let serialized = serde_json::to_string(&msg)?;
        let deserialized: McpMessage = serde_json::from_str(&serialized)?;

        assert_eq!(msg.jsonrpc, deserialized.jsonrpc);
        assert_eq!(msg.method, deserialized.method);
        assert_eq!(msg.id, deserialized.id);
        Ok(())
    }
}