add ws client and server packages

server_ws/.gitignore

@@ -0,0 +1 @@
+/target

server_ws/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "circle_server_ws"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+actix-web = "4"
+actix-web-actors = "4"
+actix = "0.13"
+env_logger = "0.10"
+log = "0.4"
+clap = { version = "4.4", features = ["derive"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+redis = { version = "0.25.0", features = ["tokio-comp"] } # For async Redis with Actix
+uuid = { version = "1.6", features = ["v4", "serde"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] } # For polling interval
+chrono = { version = "0.4", features = ["serde"] } # For timestamps
+rhai_client = { path = "/Users/timurgordon/code/git.ourworld.tf/herocode/rhaj/src/client" }
+
+[dev-dependencies]
+tokio-tungstenite = { version = "0.23.0", features = ["native-tls"] }
+futures-util = "0.3" # For StreamExt and SinkExt on WebSocket stream
+url = "2.5.0" # For parsing WebSocket URL
+circle_client_ws = { path = "../client_ws" }
+uuid = { version = "1.6", features = ["v4", "serde"] } # For e2e example, if it still uses Uuid directly for req id

server_ws/README.md

@@ -0,0 +1,52 @@
+# Circle Server WebSocket (`server_ws`)
+
+## Overview
+
+The `server_ws` component is an Actix-based WebSocket server designed to handle client connections and execute Rhai scripts. It acts as a bridge between WebSocket clients and a Rhai scripting engine, facilitating remote script execution and result retrieval.
+
+## Key Features
+
+*   **WebSocket Communication:** Establishes and manages WebSocket connections with clients.
+*   **Rhai Script Execution:** Receives Rhai scripts from clients, submits them for execution via `rhai_client`, and returns the results.
+*   **Timeout Management:** Implements timeouts for Rhai script execution to prevent indefinite blocking, returning specific error codes on timeout.
+*   **Asynchronous Processing:** Leverages Actix actors for concurrent handling of multiple client connections and script executions.
+
+## Core Components
+
+*   **`CircleWs` Actor:** The primary Actix actor responsible for handling individual WebSocket sessions. It manages the lifecycle of a client connection, processes incoming messages (Rhai scripts), and sends back results or errors.
+*   **`rhai_client` Integration:** Utilizes the `rhai_client` crate to submit scripts to a shared Rhai processing service (likely Redis-backed for task queuing and result storage) and await their completion.
+
+## Dependencies
+
+*   `actix`: Actor framework for building concurrent applications.
+*   `actix-web-actors`: WebSocket support for Actix.
+*   `rhai_client`: Client library for interacting with the Rhai scripting service.
+*   `serde_json`: For serializing and deserializing JSON messages exchanged over WebSockets.
+*   `uuid`: For generating unique task identifiers.
+
+## Workflow
+
+1.  A client establishes a WebSocket connection to the `/ws/` endpoint.
+2.  The server upgrades the connection and spawns a `CircleWs` actor instance for that session.
+3.  The client sends a JSON-RPC formatted message containing the Rhai script to be executed.
+4.  The `CircleWs` actor parses the message and uses `rhai_client::RhaiClient::submit_script_and_await_result` to send the script for execution. This method handles the interaction with the underlying task queue (e.g., Redis) and waits for the script's outcome.
+5.  The `rhai_client` will return the script's result or an error (e.g., timeout, script error).
+6.  `CircleWs` formats the result/error into a JSON-RPC response and sends it back to the client over the WebSocket.
+
+## Configuration
+
+*   **`REDIS_URL`**: The `rhai_client` component (and thus `server_ws` indirectly) relies on a Redis instance. The connection URL for this Redis instance is typically configured via an environment variable or a constant that `rhai_client` uses.
+*   **Timeout Durations**:
+    *   `TASK_TIMEOUT_DURATION` (e.g., 30 seconds): The maximum time the server will wait for a Rhai script to complete execution.
+    *   `TASK_POLL_INTERVAL_DURATION` (e.g., 200 milliseconds): The interval at which the `rhai_client` polls for task completion (this is an internal detail of `rhai_client` but relevant to understanding its behavior).
+
+## Error Handling
+
+The server implements specific JSON-RPC error responses for various scenarios:
+*   **Script Execution Timeout:** If a script exceeds `TASK_TIMEOUT_DURATION`, a specific error (e.g., code -32002) is returned.
+*   **Other `RhaiClientError`s:** Other errors originating from `rhai_client` (e.g., issues with the Redis connection, script compilation errors detected by the remote Rhai engine) are also translated into appropriate JSON-RPC error responses.
+*   **Message Parsing Errors:** Invalid incoming messages will result in error responses.
+
+## How to Run
+
+(Instructions on how to build and run the server would typically go here, e.g., `cargo run --bin circle_server_ws`)

server_ws/examples/e2e_rhai_flow.rs

@@ -0,0 +1,143 @@
+use std::process::{Command, Child, Stdio};
+use std::time::Duration;
+use std::path::PathBuf;
+use tokio::time::sleep;
+// tokio_tungstenite and direct futures_util for ws stream are no longer needed here
+// use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage};
+// use futures_util::{StreamExt, SinkExt};
+// use serde_json::Value; // No longer needed as CircleWsClient::play takes String
+// Uuid is handled by CircleWsClient internally for requests.
+// use uuid::Uuid;
+use circle_client_ws::CircleWsClient;
+// PlayResultClient and CircleWsClientError will be resolved via the client methods if needed,
+// or this indicates they were not actually needed in the scope of this file directly.
+// The compiler warning suggests they are unused from this specific import.
+
+const TEST_CIRCLE_NAME: &str = "e2e_test_circle";
+const TEST_SERVER_PORT: u16 = 9876; // Choose a unique port for the test
+const RHAI_WORKER_BIN_NAME: &str = "rhai_worker";
+const CIRCLE_SERVER_WS_BIN_NAME: &str = "circle_server_ws";
+
+// RAII guard for cleaning up child processes
+struct ChildProcessGuard {
+    child: Child,
+    name: String,
+}
+
+impl ChildProcessGuard {
+    fn new(child: Child, name: String) -> Self {
+        Self { child, name }
+    }
+}
+
+impl Drop for ChildProcessGuard {
+    fn drop(&mut self) {
+        log::info!("Cleaning up {} process (PID: {})...", self.name, self.child.id());
+        match self.child.kill() {
+            Ok(_) => {
+                log::info!("Successfully sent kill signal to {} (PID: {}).", self.name, self.child.id());
+                // Optionally wait for a short period or check status
+                match self.child.wait() {
+                    Ok(status) => log::info!("{} (PID: {}) exited with status: {}", self.name, self.child.id(), status),
+                    Err(e) => log::warn!("Error waiting for {} (PID: {}): {}", self.name, self.child.id(), e),
+                }
+            }
+            Err(e) => log::error!("Failed to kill {} (PID: {}): {}", self.name, self.child.id(), e),
+        }
+    }
+}
+
+fn find_target_dir() -> Result<PathBuf, String> {
+    // Try to find the cargo target directory relative to current exe or manifest
+    let mut current_exe = std::env::current_exe().map_err(|e| format!("Failed to get current exe path: {}", e))?;
+    // current_exe is target/debug/examples/e2e_rhai_flow
+    // want target/debug/
+    if current_exe.ends_with("examples/e2e_rhai_flow") { // Adjust if example name changes
+        current_exe.pop(); // remove e2e_rhai_flow
+        current_exe.pop(); // remove examples
+        Ok(current_exe)
+    } else {
+         // Fallback: Assume 'target/debug' relative to workspace root if CARGO_MANIFEST_DIR is set
+        let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").map_err(|_| "CARGO_MANIFEST_DIR not set".to_string())?;
+        let workspace_root = PathBuf::from(manifest_dir).parent().ok_or("Failed to get workspace root")?.to_path_buf();
+        Ok(workspace_root.join("target").join("debug"))
+    }
+}
+
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
+
+    let target_dir = find_target_dir().map_err(|e| {
+        log::error!("Could not determine target directory: {}", e);
+        e
+    })?;
+    
+    let rhai_worker_path = target_dir.join(RHAI_WORKER_BIN_NAME);
+    let circle_server_ws_path = target_dir.join(CIRCLE_SERVER_WS_BIN_NAME);
+
+    if !rhai_worker_path.exists() {
+        return Err(format!("Rhai worker binary not found at {:?}. Ensure it's built (e.g., cargo build --package rhai_worker)", rhai_worker_path).into());
+    }
+    if !circle_server_ws_path.exists() {
+        return Err(format!("Circle server WS binary not found at {:?}. Ensure it's built (e.g., cargo build --package circle_server_ws)", circle_server_ws_path).into());
+    }
+
+    log::info!("Starting {}...", RHAI_WORKER_BIN_NAME);
+    let rhai_worker_process = Command::new(&rhai_worker_path)
+        .args(["--circles", TEST_CIRCLE_NAME])
+        .stdout(Stdio::piped()) // Capture stdout
+        .stderr(Stdio::piped()) // Capture stderr
+        .spawn()?;
+    let _rhai_worker_guard = ChildProcessGuard::new(rhai_worker_process, RHAI_WORKER_BIN_NAME.to_string());
+    log::info!("{} started with PID {}", RHAI_WORKER_BIN_NAME, _rhai_worker_guard.child.id());
+
+    log::info!("Starting {} for circle '{}' on port {}...", CIRCLE_SERVER_WS_BIN_NAME, TEST_CIRCLE_NAME, TEST_SERVER_PORT);
+    let circle_server_process = Command::new(&circle_server_ws_path)
+        .args(["--port", &TEST_SERVER_PORT.to_string(), "--circle-name", TEST_CIRCLE_NAME])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()?;
+    let _circle_server_guard = ChildProcessGuard::new(circle_server_process, CIRCLE_SERVER_WS_BIN_NAME.to_string());
+    log::info!("{} started with PID {}", CIRCLE_SERVER_WS_BIN_NAME, _circle_server_guard.child.id());
+
+    // Give servers a moment to start
+    sleep(Duration::from_secs(3)).await; // Increased sleep
+
+    let ws_url_str = format!("ws://127.0.0.1:{}/ws", TEST_SERVER_PORT);
+    
+    log::info!("Creating CircleWsClient for {}...", ws_url_str);
+    let mut client = CircleWsClient::new(ws_url_str.clone());
+    
+    log::info!("Connecting CircleWsClient...");
+    client.connect().await.map_err(|e| {
+        log::error!("CircleWsClient connection failed: {}", e);
+        format!("CircleWsClient connection failed: {}", e)
+    })?;
+    log::info!("CircleWsClient connected successfully.");
+
+    let script_to_run = "let a = 5; let b = 10; print(\"E2E Rhai: \" + (a+b)); a + b";
+
+    log::info!("Sending 'play' request via CircleWsClient for script: '{}'", script_to_run);
+    
+    match client.play(script_to_run.to_string()).await {
+        Ok(play_result) => {
+            log::info!("Received play result: {:?}", play_result);
+            assert_eq!(play_result.output, "15");
+            log::info!("E2E Test Passed! Correct output '15' received via CircleWsClient.");
+        }
+        Err(e) => {
+            log::error!("CircleWsClient play request failed: {}", e);
+            return Err(format!("CircleWsClient play request failed: {}", e).into());
+        }
+    }
+    
+    log::info!("Disconnecting CircleWsClient...");
+    client.disconnect().await;
+    log::info!("CircleWsClient disconnected.");
+    
+    log::info!("E2E Rhai flow example completed successfully.");
+    // Guards will automatically clean up child processes when they go out of scope here
+    Ok(())
+}

server_ws/examples/timeout_demonstration.rs

@@ -0,0 +1,153 @@
+// Example: Timeout Demonstration for circle_server_ws
+//
+// This example demonstrates how circle_server_ws handles Rhai scripts that exceed
+// the configured execution timeout (default 30 seconds).
+//
+// This example will attempt to start its own instance of circle_server_ws.
+// Ensure circle_server_ws is compiled (cargo build --bin circle_server_ws).
+
+use circle_client_ws::CircleWsClient;
+use tokio::time::{sleep, Duration};
+use std::process::{Command, Child, Stdio};
+use std::path::PathBuf;
+
+const EXAMPLE_SERVER_PORT: u16 = 8089; // Using a specific port for this example
+const WS_URL: &str = "ws://127.0.0.1:8089/ws";
+const CIRCLE_NAME_FOR_EXAMPLE: &str = "timeout_example_circle";
+const CIRCLE_SERVER_WS_BIN_NAME: &str = "circle_server_ws";
+const SCRIPT_TIMEOUT_SECONDS: u64 = 30; // This is the server-side timeout we expect to hit
+
+// RAII guard for cleaning up child processes
+struct ChildProcessGuard {
+    child: Child,
+    name: String,
+}
+
+impl ChildProcessGuard {
+    fn new(child: Child, name: String) -> Self {
+        log::info!("{} process started with PID: {}", name, child.id());
+        Self { child, name }
+    }
+}
+
+impl Drop for ChildProcessGuard {
+    fn drop(&mut self) {
+        log::info!("Cleaning up {} process (PID: {})...", self.name, self.child.id());
+        match self.child.kill() {
+            Ok(_) => {
+                log::info!("Successfully sent kill signal to {} (PID: {}).", self.name, self.child.id());
+                match self.child.wait() {
+                    Ok(status) => log::info!("{} (PID: {}) exited with status: {}", self.name, self.child.id(), status),
+                    Err(e) => log::warn!("Error waiting for {} (PID: {}): {}", self.name, self.child.id(), e),
+                }
+            }
+            Err(e) => log::error!("Failed to kill {} (PID: {}): {}", self.name, self.child.id(), e),
+        }
+    }
+}
+
+fn find_target_bin_path(bin_name: &str) -> Result<PathBuf, String> {
+    let mut current_exe = std::env::current_exe().map_err(|e| format!("Failed to get current exe path: {}", e))?;
+    // current_exe is typically target/debug/examples/timeout_demonstration
+    // We want to find target/debug/[bin_name]
+    current_exe.pop(); // remove executable name
+    current_exe.pop(); // remove examples directory
+    let target_debug_dir = current_exe;
+    let bin_path = target_debug_dir.join(bin_name);
+    if !bin_path.exists() {
+        // Fallback: try CARGO_BIN_EXE_[bin_name] if running via `cargo run --example` which sets these
+        if let Ok(cargo_bin_path_str) = std::env::var(format!("CARGO_BIN_EXE_{}", bin_name.to_uppercase())) {
+            let cargo_bin_path = PathBuf::from(cargo_bin_path_str);
+            if cargo_bin_path.exists() {
+                return Ok(cargo_bin_path);
+            }
+        }
+        // Fallback: try target/debug/[bin_name] relative to CARGO_MANIFEST_DIR (crate root)
+        if let Ok(manifest_dir_str) = std::env::var("CARGO_MANIFEST_DIR") {
+            let bin_path_rel_manifest = PathBuf::from(manifest_dir_str).join("target").join("debug").join(bin_name);
+            if bin_path_rel_manifest.exists() {
+                return Ok(bin_path_rel_manifest);
+            }
+        }
+        return Err(format!("Binary '{}' not found at {:?}. Ensure it's built.", bin_name, bin_path));
+    }
+    Ok(bin_path)
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
+
+    let server_bin_path = find_target_bin_path(CIRCLE_SERVER_WS_BIN_NAME)?;
+    log::info!("Found server binary at: {:?}", server_bin_path);
+
+    log::info!("Starting {} for circle '{}' on port {}...", CIRCLE_SERVER_WS_BIN_NAME, CIRCLE_NAME_FOR_EXAMPLE, EXAMPLE_SERVER_PORT);
+    let server_process = Command::new(&server_bin_path)
+        .args([
+            "--port", &EXAMPLE_SERVER_PORT.to_string(),
+            "--circle-name", CIRCLE_NAME_FOR_EXAMPLE
+        ])
+        .stdout(Stdio::piped()) // Pipe stdout to keep terminal clean, or Stdio::inherit() to see server logs
+        .stderr(Stdio::piped()) // Pipe stderr as well
+        .spawn()
+        .map_err(|e| format!("Failed to start {}: {}. Ensure it is built.", CIRCLE_SERVER_WS_BIN_NAME, e))?;
+    
+    let _server_guard = ChildProcessGuard::new(server_process, CIRCLE_SERVER_WS_BIN_NAME.to_string());
+
+    log::info!("Giving the server a moment to start up...");
+    sleep(Duration::from_secs(3)).await; // Wait for server to initialize
+
+    log::info!("Attempting to connect to WebSocket server at: {}", WS_URL);
+    let mut client = CircleWsClient::new(WS_URL.to_string());
+
+    log::info!("Connecting client...");
+    if let Err(e) = client.connect().await {
+        log::error!("Failed to connect to WebSocket server: {}", e);
+        log::error!("Please check server logs if it failed to start correctly.");
+        return Err(e.into());
+    }
+    log::info!("Client connected successfully.");
+
+    // This Rhai script is designed to run for much longer than the typical server timeout.
+    let long_running_script = "
+        log(\"Rhai: Starting long-running script...\");
+        let mut x = 0;
+        for i in 0..9999999999 { // Extremely large loop
+            x = x + i;
+            if i % 100000000 == 0 {
+                // log(\"Rhai: Loop iteration \" + i);
+            }
+        }
+        // This part should not be reached if timeout works correctly.
+        log(\"Rhai: Long-running script finished calculation (x = \" + x + \").\");
+        print(x);
+        x
+    ".to_string();
+
+    log::info!("Sending long-running script (expected to time out on server after ~{}s)...", SCRIPT_TIMEOUT_SECONDS);
+
+    match client.play(long_running_script).await {
+        Ok(play_result) => {
+            log::warn!("Received unexpected success from play request: {:?}", play_result);
+            log::warn!("This might indicate the script finished faster than expected, or the timeout didn't trigger.");
+        }
+        Err(e) => {
+            log::info!("Received expected error from play request: {}", e);
+            log::info!("This demonstrates the server timing out the script execution.");
+            // You can further inspect the error details if CircleWsClientError provides them.
+            // For example, if e.to_string() contains 'code: -32002' or 'timed out'.
+            if e.to_string().contains("timed out") || e.to_string().contains("-32002") {
+                log::info!("Successfully received timeout error from the server!");
+            } else {
+                log::warn!("Received an error, but it might not be the expected timeout error: {}", e);
+            }
+        }
+    }
+
+    log::info!("Disconnecting client...");
+    client.disconnect().await;
+    log::info!("Client disconnected.");
+    log::info!("Timeout demonstration example finished.");
+
+    Ok(())
+}

server_ws/openrpc.json

@@ -0,0 +1,62 @@
+{
+  "openrpc": "1.2.6",
+  "info": {
+    "title": "Circle WebSocket Server API",
+    "version": "0.1.0",
+    "description": "API for interacting with a Circle's WebSocket server, primarily for Rhai script execution."
+  },
+  "methods": [
+    {
+      "name": "play",
+      "summary": "Executes a Rhai script on the server.",
+      "params": [
+        {
+          "name": "script",
+          "description": "The Rhai script to execute.",
+          "required": true,
+          "schema": {
+            "type": "string"
+          }
+        }
+      ],
+      "result": {
+        "name": "playResult",
+        "description": "The output from the executed Rhai script.",
+        "schema": {
+          "$ref": "#/components/schemas/PlayResult"
+        }
+      },
+      "examples": [
+        {
+          "name": "Simple Script Execution",
+          "params": [
+            {
+              "name": "script",
+              "value": "let x = 10; x * 2"
+            }
+          ],
+          "result": {
+            "name": "playResult",
+            "value": {
+              "output": "20"
+            }
+          }
+        }
+      ]
+    }
+  ],
+  "components": {
+    "schemas": {
+      "PlayResult": {
+        "type": "object",
+        "properties": {
+          "output": {
+            "type": "string",
+            "description": "The string representation of the Rhai script's evaluation result."
+          }
+        },
+        "required": ["output"]
+      }
+    }
+  }
+}

server_ws/src/main.rs

@@ -0,0 +1,260 @@
+use actix_web::{web, App, HttpRequest, HttpServer, HttpResponse, Error};
+use actix_web_actors::ws;
+use actix::{Actor, ActorContext, StreamHandler, AsyncContext, WrapFuture, ActorFutureExt};
+// HashMap no longer needed
+use clap::Parser;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::time::Duration;
+// AsyncCommands no longer directly used here
+use rhai_client::RhaiClientError; // Import RhaiClientError for matching
+// Uuid is not directly used here anymore for task_id generation, RhaiClient handles it.
+// Utc no longer directly used here
+// RhaiClientError is not directly handled here, errors from RhaiClient are strings or RhaiClient's own error type.
+use rhai_client::RhaiClient; // ClientRhaiTaskDetails is used via rhai_client::RhaiTaskDetails
+
+const REDIS_URL: &str = "redis://127.0.0.1/"; // Make this configurable if needed
+
+// JSON-RPC 2.0 Structures
+#[derive(Serialize, Deserialize, Debug, Clone)] // Added Clone
+struct JsonRpcRequest {
+    jsonrpc: String,
+    method: String,
+    params: Value,
+    id: Option<Value>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)] // Added Clone
+struct JsonRpcResponse {
+    jsonrpc: String,
+    result: Option<Value>,
+    error: Option<JsonRpcError>,
+    id: Value,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)] // Added Clone
+struct JsonRpcError {
+    code: i32,
+    message: String,
+    data: Option<Value>,
+}
+
+// Specific params and result for "play" method
+#[derive(Serialize, Deserialize, Debug, Clone)] // Added Clone
+struct PlayParams {
+    script: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)] // Added Clone
+struct PlayResult {
+    output: String,
+}
+
+// Local RhaiTaskDetails struct is removed, will use ClientRhaiTaskDetails from rhai_client crate.
+// Ensure field names used in polling logic (e.g. error_message) are updated if they differ.
+// rhai_client::RhaiTaskDetails uses 'error' and 'client_rpc_id'.
+
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+struct Args {
+    #[clap(short, long, value_parser, default_value_t = 8080)]
+    port: u16,
+
+    #[clap(short, long, value_parser)]
+    circle_name: String,
+}
+
+// WebSocket Actor
+struct CircleWs {
+    server_circle_name: String,
+    // redis_client field removed as RhaiClient handles its own connection
+}
+
+const TASK_TIMEOUT_DURATION: Duration = Duration::from_secs(30); // 30 seconds timeout
+const TASK_POLL_INTERVAL_DURATION: Duration = Duration::from_millis(200); // 200 ms poll interval
+
+impl CircleWs {
+    fn new(name: String) -> Self {
+        Self {
+            server_circle_name: name,
+        }
+    }
+}
+
+impl Actor for CircleWs {
+    type Context = ws::WebsocketContext<Self>;
+
+    fn started(&mut self, _ctx: &mut Self::Context) {
+        log::info!("WebSocket session started for server dedicated to: {}", self.server_circle_name);
+    }
+
+    fn stopping(&mut self, _ctx: &mut Self::Context) -> actix::Running {
+        log::info!("WebSocket session stopping for server dedicated to: {}", self.server_circle_name);
+        actix::Running::Stop
+    }
+}
+
+// WebSocket message handler
+impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for CircleWs {
+    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
+        match msg {
+            Ok(ws::Message::Text(text)) => {
+                log::info!("WS Text for {}: {}", self.server_circle_name, text);
+                match serde_json::from_str::<JsonRpcRequest>(&text) {
+                    Ok(req) => {
+                        let client_rpc_id = req.id.clone().unwrap_or(Value::Null);
+                        if req.method == "play" {
+                            match serde_json::from_value::<PlayParams>(req.params.clone()) {
+                                Ok(play_params) => {
+                                    // Use RhaiClient to submit the script
+                                    let script_content = play_params.script;
+                                    let current_circle_name = self.server_circle_name.clone();
+                                    let rpc_id_for_client = client_rpc_id.clone(); // client_rpc_id is already Value
+
+                                    let fut = async move {
+                                        match RhaiClient::new(REDIS_URL) {
+                                            Ok(rhai_task_client) => {
+                                                rhai_task_client.submit_script_and_await_result(
+                                                    &current_circle_name,
+                                                    script_content,
+                                                    Some(rpc_id_for_client.clone()),
+                                                    TASK_TIMEOUT_DURATION,
+                                                    TASK_POLL_INTERVAL_DURATION,
+                                                ).await // This returns Result<rhai_client::RhaiTaskDetails, RhaiClientError>
+                                            }
+                                            Err(e) => {
+                                                log::error!("Failed to create RhaiClient: {}", e);
+                                                Err(e) // Convert the error from RhaiClient::new into the type expected by the map function's error path.
+                                            }
+                                        }
+                                    };
+
+                                    ctx.spawn(fut.into_actor(self).map(move |result, _act, ws_ctx| {
+                                        let response = match result {
+                                            Ok(task_details) => { // ClientRhaiTaskDetails
+                                                if task_details.status == "completed" {
+                                                    log::info!("Task completed successfully. Client RPC ID: {:?}, Output: {:?}", task_details.client_rpc_id, task_details.output);
+                                                    JsonRpcResponse {
+                                                        jsonrpc: "2.0".to_string(),
+                                                        result: Some(serde_json::to_value(PlayResult { 
+                                                            output: task_details.output.unwrap_or_default() 
+                                                        }).unwrap()),
+                                                        error: None,
+                                                        id: client_rpc_id, // Use the original client_rpc_id from the request
+                                                    }
+                                                } else { // status == "error"
+                                                    log::warn!("Task execution failed. Client RPC ID: {:?}, Error: {:?}", task_details.client_rpc_id, task_details.error);
+                                                    JsonRpcResponse {
+                                                        jsonrpc: "2.0".to_string(),
+                                                        result: None,
+                                                        error: Some(JsonRpcError {
+                                                            code: -32004, // Script execution error
+                                                            message: task_details.error.unwrap_or_else(|| "Script execution failed with no specific error message".to_string()),
+                                                            data: None,
+                                                        }),
+                                                        id: client_rpc_id,
+                                                    }
+                                                }
+                                            }
+                                            Err(rhai_err) => { // RhaiClientError
+                                                log::error!("RhaiClient operation failed: {}", rhai_err);
+                                                let (code, message) = match rhai_err {
+                                                    RhaiClientError::Timeout(task_id) => (-32002, format!("Timeout waiting for task {} to complete", task_id)),
+                                                    RhaiClientError::RedisError(e) => (-32003, format!("Redis communication error: {}", e)),
+                                                    RhaiClientError::SerializationError(e) => (-32003, format!("Serialization error: {}", e)),
+                                                    RhaiClientError::TaskNotFound(task_id) => (-32005, format!("Task {} not found after submission", task_id)),
+                                                };
+                                                JsonRpcResponse {
+                                                    jsonrpc: "2.0".to_string(),
+                                                    result: None,
+                                                    id: client_rpc_id,
+                                                    error: Some(JsonRpcError { code, message, data: None }),
+                                                }
+                                            }
+                                        };
+                                        ws_ctx.text(serde_json::to_string(&response).unwrap());
+                                    }));
+                                }
+                                Err(e) => { // Invalid params for 'play'
+                                    log::error!("Invalid params for 'play' method: {}", e);
+                                    let err_resp = JsonRpcResponse {
+                                        jsonrpc: "2.0".to_string(), result: None, id: client_rpc_id,
+                                        error: Some(JsonRpcError { code: -32602, message: "Invalid params".to_string(), data: Some(Value::String(e.to_string())) }),
+                                    };
+                                    ctx.text(serde_json::to_string(&err_resp).unwrap());
+                                }
+                            }
+                        } else { // Method not found
+                            log::warn!("Method not found: {}", req.method);
+                             let err_resp = JsonRpcResponse {
+                                jsonrpc: "2.0".to_string(), result: None, id: client_rpc_id,
+                                error: Some(JsonRpcError { code: -32601, message: "Method not found".to_string(), data: None }),
+                            };
+                            ctx.text(serde_json::to_string(&err_resp).unwrap());
+                        }
+                    }
+                    Err(e) => { // Parse error
+                        log::error!("Failed to parse JSON-RPC request: {}", e);
+                        let err_resp = JsonRpcResponse {
+                            jsonrpc: "2.0".to_string(), result: None, id: Value::Null, // No ID if request couldn't be parsed
+                            error: Some(JsonRpcError { code: -32700, message: "Parse error".to_string(), data: Some(Value::String(e.to_string())) }),
+                        };
+                        ctx.text(serde_json::to_string(&err_resp).unwrap());
+                    }
+                }
+            }
+            Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
+            Ok(ws::Message::Pong(_)) => {},
+            Ok(ws::Message::Binary(_bin)) => log::warn!("Binary messages not supported."),
+            Ok(ws::Message::Close(reason)) => {
+                ctx.close(reason);
+                ctx.stop();
+            }
+            Ok(ws::Message::Continuation(_)) => ctx.stop(),
+            Ok(ws::Message::Nop) => (),
+            Err(e) => {
+                log::error!("WS Error: {:?}", e);
+                ctx.stop();
+            }
+        }
+    }
+}
+
+// WebSocket handshake and actor start
+async fn ws_handler(
+    req: HttpRequest,
+    stream: web::Payload,
+    server_name: web::Data<String>,
+    // redis_client: web::Data<redis::Client>, // No longer passed to CircleWs actor directly
+) -> Result<HttpResponse, Error> {
+    log::info!("WebSocket handshake attempt for server: {}", server_name.get_ref());
+    let resp = ws::start(
+        CircleWs::new(server_name.get_ref().clone()), // Pass only the server name
+        &req,
+        stream
+    )?;
+    Ok(resp)
+}
+
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+    let args = Args::parse();
+    
+    std::env::set_var("RUST_LOG", "info,circle_server_ws=debug");
+    env_logger::init();
+    
+    log::info!(
+        "Starting WebSocket server for Circle: '{}' on port {}...",
+        args.circle_name, args.port
+    );
+
+    HttpServer::new(move || {
+        App::new()
+            .app_data(web::Data::new(args.circle_name.clone()))
+            .route("/ws", web::get().to(ws_handler))
+            .default_service(web::route().to(|| async { HttpResponse::NotFound().body("404 Not Found - This is a WebSocket-only server for a specific circle.") }))
+    })
+    .bind(("127.0.0.1", args.port))?
+    .run()
+    .await
+}

server_ws/tests/timeout_integration_test.rs

@@ -0,0 +1,135 @@
+use tokio::time::Duration; // Removed unused sleep
+use futures_util::{sink::SinkExt, stream::StreamExt};
+use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
+use serde_json::Value; // Removed unused json macro import
+use std::process::Command;
+use std::thread;
+use std::sync::Once;
+
+// Define a simple JSON-RPC request structure for sending scripts
+#[derive(serde::Serialize, Debug)]
+struct JsonRpcRequest {
+    jsonrpc: String,
+    method: String,
+    params: ScriptParams,
+    id: u64,
+}
+
+#[derive(serde::Serialize, Debug)]
+struct ScriptParams {
+    script: String,
+}
+
+// Define a simple JSON-RPC error response structure for assertion
+#[derive(serde::Deserialize, Debug)]
+struct JsonRpcErrorResponse {
+    _jsonrpc: String, // Field is present in response, but not used in assert
+    error: JsonRpcErrorDetails,
+    _id: Option<Value>, // Field is present in response, but not used in assert
+}
+
+#[derive(serde::Deserialize, Debug)]
+struct JsonRpcErrorDetails {
+    code: i32,
+    message: String,
+}
+
+const SERVER_ADDRESS: &str = "ws://127.0.0.1:8088/ws"; // Match port in main.rs or make configurable
+const TEST_CIRCLE_NAME: &str = "test_timeout_circle";
+const SERVER_STARTUP_TIME: Duration = Duration::from_secs(5); // Time to wait for server to start
+const RHAI_TIMEOUT_SECONDS: u64 = 30; // Should match TASK_TIMEOUT_DURATION in circle_server_ws
+
+static START_SERVER: Once = Once::new();
+
+fn ensure_server_is_running() {
+    START_SERVER.call_once(|| {
+        println!("Attempting to start circle_server_ws for integration tests...");
+        // The server executable will be in target/debug relative to the crate root
+        let server_executable = "target/debug/circle_server_ws"; 
+        
+        thread::spawn(move || {
+            let mut child = Command::new(server_executable)
+                .arg("--port=8088") // Use a specific port for testing
+                .arg(format!("--circle-name={}", TEST_CIRCLE_NAME))
+                .spawn()
+                .expect("Failed to start circle_server_ws. Make sure it's compiled (cargo build).");
+            
+            let status = child.wait().expect("Failed to wait on server process.");
+            println!("Server process exited with status: {}", status);
+        });
+        println!("Server start command issued. Waiting for {}s...", SERVER_STARTUP_TIME.as_secs());
+        thread::sleep(SERVER_STARTUP_TIME);
+        println!("Presumed server started.");
+    });
+}
+
+#[tokio::test]
+async fn test_rhai_script_timeout() {
+    ensure_server_is_running();
+
+    println!("Connecting to WebSocket server: {}", SERVER_ADDRESS);
+    let (mut ws_stream, _response) = connect_async(SERVER_ADDRESS)
+        .await
+        .expect("Failed to connect to WebSocket server");
+    println!("Connected to WebSocket server.");
+
+    // Rhai script designed to run longer than RHAI_TIMEOUT_SECONDS
+    // A large loop should cause a timeout.
+    let long_running_script = format!("
+        let mut x = 0;
+        for i in 0..999999999 {{
+            x = x + i;
+            if i % 10000000 == 0 {{
+                // debug(\"Looping: \" + i); // Optional: for server-side logging if enabled
+            }}
+        }}
+        print(x); // This line will likely not be reached due to timeout
+    ");
+
+    let request = JsonRpcRequest {
+        jsonrpc: "2.0".to_string(),
+        method: "execute_script".to_string(),
+        params: ScriptParams { script: long_running_script },
+        id: 1,
+    };
+
+    let request_json = serde_json::to_string(&request).expect("Failed to serialize request");
+    println!("Sending long-running script request: {}", request_json);
+    ws_stream.send(Message::Text(request_json)).await.expect("Failed to send message");
+
+    println!("Waiting for response (expecting timeout after ~{}s)..", RHAI_TIMEOUT_SECONDS);
+
+    // Wait for a response, expecting a timeout error
+    // The server's timeout is RHAI_TIMEOUT_SECONDS, client should wait a bit longer.
+    match tokio::time::timeout(Duration::from_secs(RHAI_TIMEOUT_SECONDS + 15), ws_stream.next()).await {
+        Ok(Some(Ok(Message::Text(text)))) => {
+            println!("Received response: {}", text);
+            let response: Result<JsonRpcErrorResponse, _> = serde_json::from_str(&text);
+            match response {
+                Ok(err_resp) => {
+                    assert_eq!(err_resp.error.code, -32002, "Error code should indicate timeout.");
+                    assert!(err_resp.error.message.contains("timed out"), "Error message should indicate timeout.");
+                    println!("Timeout test passed! Received correct timeout error.");
+                }
+                Err(e) => {
+                    panic!("Failed to deserialize error response: {}. Raw: {}", e, text);
+                }
+            }
+        }
+        Ok(Some(Ok(other_msg))) => {
+            panic!("Received unexpected message type: {:?}", other_msg);
+        }
+        Ok(Some(Err(e))) => {
+            panic!("WebSocket error: {}", e);
+        }
+        Ok(None) => {
+            panic!("WebSocket stream closed unexpectedly.");
+        }
+        Err(_) => {
+            panic!("Test timed out waiting for server response. Server might not have sent timeout error or took too long.");
+        }
+    }
+
+    ws_stream.close(None).await.ok();
+    println!("Test finished, WebSocket closed.");
+}