Merge pull request 'Fix smoke tests and resolve CI timeouts' (#1) from mcp-integration into main

Reviewed-on: #1

Cargo.lock

@@ -1389,7 +1389,9 @@ dependencies = [
  "crossterm 0.28.1",
  "env_logger",
  "log",
+ "rand",
  "ratatui",
+ "regex",
  "rusqlite",
  "russh",
  "serde",

Cargo.toml

@@ -14,3 +14,5 @@ ratatui = "0.30"
 crossterm = "0.28"
 log = "0.4"
 env_logger = "0.11"
+regex = "1"
+rand = "0.8"

run-tests.sh

@@ -31,6 +31,8 @@ ssh_mud smoketest@localhost <<'EOF'
 1
 look
 stats
+go south
+go down
 go north
 talk barkeep
 go south
@@ -73,17 +75,18 @@ EOF
 # Test 6: Tick-based combat (connect and wait for ticks)
 ./target/debug/mudtool -d "$TEST_DB" settings set registration_open true
 ./target/debug/mudtool -d "$TEST_DB" players delete smoketest
-ssh_mud smoketest@localhost <<'EOF'
+# Use subshell to pipe commands with a delay between them while staying connected
-1
+(
-1
+  echo "1"
-go south
+  echo "1"
-attack thief
+  echo "go south"
-EOF
+  echo "go down"
+  echo "go south"
+  echo "attack thief"
   sleep 8
-ssh_mud smoketest@localhost <<'EOF'
+  echo "stats"
-stats
+  echo "quit"
-quit
+) | ssh_mud smoketest@localhost
-EOF
 
 # Cleanup (trap handles server kill)
 ./target/debug/mudtool -d "$TEST_DB" settings set registration_open true

src/admin.rs

@@ -89,14 +89,15 @@ async fn admin_promote(target: &str, state: &SharedState) -> CommandResult {
                     )
                     .as_bytes(),
                 );
+                if let (Some(ch), Some(h)) = (conn.channel, &conn.handle) {
                     return CommandResult {
                         output: format!(
                             "{}\r\n",
                             ansi::system_msg(&format!("{target} has been promoted to admin."))
                         ),
                         broadcasts: vec![BroadcastMsg {
-                        channel: conn.channel,
+                            channel: ch,
-                        handle: conn.handle.clone(),
+                            handle: h.clone(),
                             data: msg,
                         }],
                         kick_targets: Vec::new(),
@@ -104,6 +105,7 @@ async fn admin_promote(target: &str, state: &SharedState) -> CommandResult {
                     };
                 }
             }
+        }
         simple(&format!(
             "{}\r\n",
             ansi::system_msg(&format!(
@@ -188,18 +190,31 @@ async fn admin_kick(target: &str, player_id: usize, state: &SharedState) -> Comm
             let mut bcast: Vec<BroadcastMsg> = st
                 .players_in_room(&room_id, player_id)
                 .iter()
-                .map(|p| BroadcastMsg {
+                .filter_map(|p| {
-                    channel: p.channel,
+                    if let (Some(ch), Some(h)) = (p.channel, &p.handle) {
-                    handle: p.handle.clone(),
+                        Some(BroadcastMsg {
+                            channel: ch,
+                            handle: h.clone(),
                             data: departure.clone(),
                         })
+                    } else {
+                        None
+                    }
+                })
                 .collect();
             // Send kick message to the target before closing
+            let mut kick_targets = Vec::new();
+            if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
                 bcast.push(BroadcastMsg {
-                channel: c.channel,
+                    channel: ch,
-                handle: c.handle.clone(),
+                    handle: h.clone(),
                     data: kick_msg,
                 });
+                kick_targets.push(KickTarget {
+                    channel: ch,
+                    handle: h.clone(),
+                });
+            }
 
             CommandResult {
                 output: format!(
@@ -207,10 +222,7 @@ async fn admin_kick(target: &str, player_id: usize, state: &SharedState) -> Comm
                     ansi::system_msg(&format!("Kicked {name} from the server."))
                 ),
                 broadcasts: bcast,
-                kick_targets: vec![KickTarget {
+                kick_targets,
-                    channel: c.channel,
-                    handle: c.handle.clone(),
-                }],
                 quit: false,
             }
         }
@@ -265,11 +277,17 @@ async fn admin_teleport(room_id: &str, player_id: usize, state: &SharedState) ->
     let mut bcast: Vec<BroadcastMsg> = st
         .players_in_room(&old_rid, player_id)
         .iter()
-        .map(|c| BroadcastMsg {
+        .filter_map(|c| {
-            channel: c.channel,
+            if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
-            handle: c.handle.clone(),
+                Some(BroadcastMsg {
+                    channel: ch,
+                    handle: h.clone(),
                     data: leave.clone(),
                 })
+            } else {
+                None
+            }
+        })
         .collect();
 
     if let Some(c) = st.players.get_mut(&player_id) {
@@ -286,12 +304,14 @@ async fn admin_teleport(room_id: &str, player_id: usize, state: &SharedState) ->
         .as_bytes(),
     );
     for c in st.players_in_room(room_id, player_id) {
+        if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
             bcast.push(BroadcastMsg {
-            channel: c.channel,
+                channel: ch,
-            handle: c.handle.clone(),
+                handle: h.clone(),
                 data: arrive.clone(),
             });
         }
+    }
 
     st.save_player_to_db(player_id);
     let view = crate::commands::render_room_view(room_id, player_id, &st);
@@ -360,11 +380,17 @@ async fn admin_announce(msg: &str, player_id: usize, state: &SharedState) -> Com
         .players
         .iter()
         .filter(|(&id, _)| id != player_id)
-        .map(|(_, c)| BroadcastMsg {
+        .filter_map(|(_, c)| {
-            channel: c.channel,
+            if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
-            handle: c.handle.clone(),
+                Some(BroadcastMsg {
+                    channel: ch,
+                    handle: h.clone(),
                     data: announcement.clone(),
                 })
+            } else {
+                None
+            }
+        })
         .collect();
 
     CommandResult {
@@ -408,6 +434,8 @@ async fn admin_heal(args: &str, player_id: usize, state: &SharedState) -> Comman
                 c.player.stats.hp = c.player.stats.max_hp;
                 let name = c.player.name.clone();
                 let hp = c.player.stats.max_hp;
+                let mut bcast = Vec::new();
+                if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
                     let notify = CryptoVec::from(
                         format!(
                             "\r\n{}\r\n{}",
@@ -416,11 +444,12 @@ async fn admin_heal(args: &str, player_id: usize, state: &SharedState) -> Comman
                         )
                         .as_bytes(),
                     );
-                let bcast = vec![BroadcastMsg {
+                    bcast.push(BroadcastMsg {
-                    channel: c.channel,
+                        channel: ch,
-                    handle: c.handle.clone(),
+                        handle: h.clone(),
                         data: notify,
-                }];
+                    });
+                }
                 let _ = c;
                 st.save_player_to_db(tid);
                 return CommandResult {

src/commands.rs

@@ -42,17 +42,42 @@ fn resolve_dir(input: &str) -> &str {
     input
 }
 
-pub async fn execute(
+pub async fn execute_for_ssh(
     input: &str,
     player_id: usize,
     state: &SharedState,
     session: &mut Session,
     channel: ChannelId,
 ) -> Result<bool, russh::Error> {
+    let result = execute(input, player_id, state).await;
+
+    send(session, channel, &result.output)?;
+    for msg in result.broadcasts {
+        let _ = msg.handle.data(msg.channel, msg.data).await;
+    }
+    for kick in result.kick_targets {
+        let _ = kick.handle.close(kick.channel).await;
+    }
+    if result.quit {
+        return Ok(false);
+    }
+    send(session, channel, &ansi::prompt())?;
+    Ok(true)
+}
+
+pub async fn execute(
+    input: &str,
+    player_id: usize,
+    state: &SharedState,
+) -> CommandResult {
     let input = input.trim();
     if input.is_empty() {
-        send(session, channel, &ansi::prompt())?;
+        return CommandResult {
-        return Ok(true);
+            output: ansi::prompt(),
+            broadcasts: Vec::new(),
+            kick_targets: Vec::new(),
+            quit: false,
+        };
     }
 
     let (cmd, args) = match input.split_once(' ') {
@@ -72,24 +97,23 @@ pub async fn execute(
                         | "spells" | "skills" | "quit" | "exit"
                 )
             {
-                drop(st);
+                return CommandResult {
-                send(
+                    output: format!(
-                    session,
-                    channel,
-                    &format!(
                         "{}\r\n{}",
                         ansi::error_msg(
                             "You're in combat! Use 'attack', 'defend', 'flee', 'cast', 'use', 'look', 'stats', or 'inventory'."
                         ),
                         ansi::prompt()
                     ),
-                )?;
+                    broadcasts: Vec::new(),
-                return Ok(true);
+                    kick_targets: Vec::new(),
+                    quit: false,
+                };
             }
         }
     }
 
-    let result = match cmd.as_str() {
+    match cmd.as_str() {
         "look" | "l" => cmd_look(player_id, &args, state).await,
         "go" => cmd_go(player_id, &args, state).await,
         "north" | "south" | "east" | "west" | "up" | "down" | "n" | "s" | "e" | "w" | "u"
@@ -118,27 +142,20 @@ pub async fn execute(
             kick_targets: Vec::new(),
             quit: true,
         },
-        _ => simple(&format!(
+        _ => CommandResult {
+            output: format!(
                 "{}\r\n",
                 ansi::error_msg(&format!(
                     "Unknown command: '{cmd}'. Type 'help' for commands."
                 ))
-        )),
+            ),
-    };
+            broadcasts: Vec::new(),
+            kick_targets: Vec::new(),
+            quit: false,
+        },
+    }
+}
 
-    send(session, channel, &result.output)?;
-    for msg in result.broadcasts {
-        let _ = msg.handle.data(msg.channel, msg.data).await;
-    }
-    for kick in result.kick_targets {
-        let _ = kick.handle.close(kick.channel).await;
-    }
-    if result.quit {
-        return Ok(false);
-    }
-    send(session, channel, &ansi::prompt())?;
-    Ok(true)
-}
 
 fn send(session: &mut Session, channel: ChannelId, text: &str) -> Result<(), russh::Error> {
     session.data(channel, CryptoVec::from(text.as_bytes()))?;
@@ -457,12 +474,14 @@ async fn cmd_go(pid: usize, direction: &str, state: &SharedState) -> CommandResu
     );
     let mut bcast = Vec::new();
     for c in st.players_in_room(&old_rid, pid) {
+        if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
             bcast.push(BroadcastMsg {
-            channel: c.channel,
+                channel: ch,
-            handle: c.handle.clone(),
+                handle: h.clone(),
                 data: leave.clone(),
             });
         }
+    }
 
     if let Some(c) = st.players.get_mut(&pid) {
         c.player.room_id = new_rid.clone();
@@ -477,12 +496,14 @@ async fn cmd_go(pid: usize, direction: &str, state: &SharedState) -> CommandResu
         .as_bytes(),
     );
     for c in st.players_in_room(&new_rid, pid) {
+        if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
             bcast.push(BroadcastMsg {
-            channel: c.channel,
+                channel: ch,
-            handle: c.handle.clone(),
+                handle: h.clone(),
                 data: arrive.clone(),
             });
         }
+    }
 
     st.save_player_to_db(pid);
     let output = render_room_view(&new_rid, pid, &st);
@@ -524,11 +545,17 @@ async fn cmd_say(pid: usize, msg: &str, state: &SharedState) -> CommandResult {
     let bcast: Vec<_> = st
         .players_in_room(&rid, pid)
         .iter()
-        .map(|c| BroadcastMsg {
+        .filter_map(|c| {
-            channel: c.channel,
+            if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
-            handle: c.handle.clone(),
+                Some(BroadcastMsg {
+                    channel: ch,
+                    handle: h.clone(),
                     data: other.clone(),
                 })
+            } else {
+                None
+            }
+        })
         .collect();
     CommandResult {
         output: self_msg,

src/game.rs

@@ -98,8 +98,8 @@ pub struct NpcInstance {
 
 pub struct PlayerConnection {
     pub player: Player,
-    pub channel: ChannelId,
+    pub channel: Option<ChannelId>,
-    pub handle: Handle,
+    pub handle: Option<Handle>,
     pub combat: Option<CombatState>,
 }
 
@@ -277,8 +277,8 @@ impl GameState {
         name: String,
         race_id: String,
         class_id: String,
-        channel: ChannelId,
+        channel: Option<ChannelId>,
-        handle: Handle,
+        handle: Option<Handle>,
     ) {
         let room_id = self.world.spawn_room.clone();
         let race = self.world.races.iter().find(|r| r.id == race_id);
@@ -356,8 +356,8 @@ impl GameState {
         &mut self,
         id: usize,
         saved: SavedPlayer,
-        channel: ChannelId,
+        channel: Option<ChannelId>,
-        handle: Handle,
+        handle: Option<Handle>,
     ) {
         let inventory: Vec<Object> =
             serde_json::from_str(&saved.inventory_json).unwrap_or_default();

src/jsonrpc.rs

@@ -0,0 +1,172 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::Mutex;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::game::SharedState;
+use crate::commands;
+
+#[derive(Deserialize)]
+struct JsonRpcRequest {
+    jsonrpc: String,
+    method: String,
+    params: Option<serde_json::Value>,
+    id: Option<serde_json::Value>,
+}
+
+#[derive(Serialize)]
+struct JsonRpcResponse {
+    jsonrpc: String,
+    result: Option<serde_json::Value>,
+    error: Option<serde_json::Value>,
+    id: Option<serde_json::Value>,
+}
+
+pub async fn run_jsonrpc_server(state: SharedState, port: u16) {
+    let listener = TcpListener::bind(("0.0.0.0", port)).await.unwrap();
+    log::info!("JSON-RPC server listening on 0.0.0.0:{port}");
+
+    let sessions = Arc::new(Mutex::new(HashMap::<usize, String>::new()));
+
+    loop {
+        let (stream, addr) = match listener.accept().await {
+            Ok(s) => s,
+            Err(e) => {
+                log::error!("Failed to accept connection: {e}");
+                continue;
+            }
+        };
+        log::info!("New JSON-RPC connection from {addr:?}");
+
+        let state = state.clone();
+        let sessions = sessions.clone();
+        tokio::spawn(async move {
+            handle_connection(stream, state, sessions).await;
+        });
+    }
+}
+
+async fn handle_connection(
+    mut stream: TcpStream,
+    state: SharedState,
+    sessions: Arc<Mutex<HashMap<usize, String>>>,
+) {
+    let (reader, mut writer) = stream.split();
+    let mut reader = BufReader::new(reader);
+    let mut line = String::new();
+
+    // Map RPC session ID to player ID
+    let mut current_player_id: Option<usize> = None;
+
+    loop {
+        line.clear();
+        match reader.read_line(&mut line).await {
+            Ok(0) => break, // Connection closed
+            Ok(_) => {
+                let req: JsonRpcRequest = match serde_json::from_str(&line) {
+                    Ok(r) => r,
+                    Err(e) => {
+                        let resp = JsonRpcResponse {
+                            jsonrpc: "2.0".to_string(),
+                            result: None,
+                            error: Some(json!({"code": -32700, "message": format!("Parse error: {e}")})),
+                            id: None,
+                        };
+                        let _ = writer.write_all(format!("{}\n", serde_json::to_string(&resp).unwrap()).as_bytes()).await;
+                        continue;
+                    }
+                };
+
+                let resp = handle_request(req, &state, &sessions, &mut current_player_id).await;
+                let _ = writer.write_all(format!("{}\n", serde_json::to_string(&resp).unwrap()).as_bytes()).await;
+            }
+            Err(e) => {
+                log::error!("Error reading from JSON-RPC stream: {e}");
+                break;
+            }
+        }
+    }
+
+    // Cleanup session if needed
+    if let Some(pid) = current_player_id {
+        let mut st = state.lock().await;
+        st.remove_player(pid);
+    }
+}
+
+async fn handle_request(
+    req: JsonRpcRequest,
+    state: &SharedState,
+    _sessions: &Arc<Mutex<HashMap<usize, String>>>,
+    current_player_id: &mut Option<usize>,
+) -> JsonRpcResponse {
+    let method = req.method.as_str();
+    let id = req.id.clone();
+
+    let result = match method {
+        "login" => {
+            let username = req.params.as_ref()
+                .and_then(|p| p.get("username"))
+                .and_then(|u| u.as_str())
+                .unwrap_or("anonymous");
+            
+            let mut st = state.lock().await;
+            let player_id = rand::random::<usize>();
+            
+            let saved = st.db.load_player(username);
+            if let Some(saved) = saved {
+                st.load_existing_player(player_id, saved, None, None);
+                *current_player_id = Some(player_id);
+                json!({"status": "success", "session_id": player_id})
+            } else {
+                json!({"status": "error", "message": "Player not found. Create character via SSH first."})
+            }
+        },
+        "list_commands" => {
+            json!([
+                "look", "go", "north", "south", "east", "west", "up", "down",
+                "say", "who", "take", "drop", "inventory", "equip", "use",
+                "examine", "talk", "attack", "defend", "flee", "cast",
+                "spells", "skills", "guild", "stats", "help"
+            ])
+        },
+        "execute" => {
+            if let Some(pid) = *current_player_id {
+                let command = req.params.as_ref()
+                    .and_then(|p| p.get("command"))
+                    .and_then(|c| c.as_str())
+                    .unwrap_or("");
+                let args = req.params.as_ref()
+                    .and_then(|p| p.get("args"))
+                    .and_then(|a| a.as_str())
+                    .unwrap_or("");
+                
+                let input = if args.is_empty() { command.to_string() } else { format!("{command} {args}") };
+                let result = commands::execute(&input, pid, state).await;
+                
+                json!({
+                    "output": strip_ansi(&result.output),
+                    "quit": result.quit
+                })
+            } else {
+                json!({"error": "Not logged in"})
+            }
+        },
+        _ => json!({"error": "Method not found"}),
+    };
+
+    JsonRpcResponse {
+        jsonrpc: "2.0".to_string(),
+        result: Some(result),
+        error: None,
+        id,
+    }
+}
+
+fn strip_ansi(text: &str) -> String {
+    let re = regex::Regex::new(r"\x1b\[[0-9;]*[a-zA-Z]").unwrap();
+    re.replace_all(text, "").to_string()
+}

src/lib.rs

@@ -8,3 +8,4 @@ pub mod game;
 pub mod ssh;
 pub mod tick;
 pub mod world;
+pub mod jsonrpc;

src/main.rs

@@ -21,6 +21,7 @@ async fn main() {
     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
 
     let mut port = DEFAULT_PORT;
+    let mut jsonrpc_port = 2223;
     let mut world_dir = PathBuf::from(DEFAULT_WORLD_DIR);
     let mut db_path = PathBuf::from(DEFAULT_DB_PATH);
 
@@ -35,6 +36,13 @@ async fn main() {
                     .and_then(|s| s.parse().ok())
                     .expect("--port requires a number");
             }
+            "--rpc-port" => {
+                i += 1;
+                jsonrpc_port = args
+                    .get(i)
+                    .and_then(|s| s.parse().ok())
+                    .expect("--rpc-port requires a number");
+            }
             "--world" | "-w" => {
                 i += 1;
                 world_dir = PathBuf::from(args.get(i).expect("--world requires a path"));
@@ -45,7 +53,8 @@ async fn main() {
             }
             "--help" => {
                 eprintln!("Usage: mudserver [OPTIONS]");
-                eprintln!("  --port, -p   Listen port (default: {DEFAULT_PORT})");
+                eprintln!("  --port, -p      SSH listen port (default: {DEFAULT_PORT})");
+                eprintln!("  --rpc-port      JSON-RPC listen port (default: 2223)");
                 eprintln!("  --world, -w     World directory (default: {DEFAULT_WORLD_DIR})");
                 eprintln!("  --db, -d        Database path (default: {DEFAULT_DB_PATH})");
                 std::process::exit(0);
@@ -90,6 +99,12 @@ async fn main() {
         tick::run_tick_engine(tick_state).await;
     });
 
+    // Spawn JSON-RPC server
+    let rpc_state = state.clone();
+    tokio::spawn(async move {
+        mudserver::jsonrpc::run_jsonrpc_server(rpc_state, jsonrpc_port).await;
+    });
+
     let mut server = ssh::MudServer::new(state);
 
     let listener = TcpListener::bind(("0.0.0.0", port)).await.unwrap();

src/ssh.rs

@@ -80,7 +80,7 @@ impl MudHandler {
         if let Some(saved) = saved {
             let handle = session.handle();
             let mut state = self.state.lock().await;
-            state.load_existing_player(self.id, saved, channel, handle);
+            state.load_existing_player(self.id, saved, Some(channel), Some(handle));
             drop(state);
 
             let msg = format!(
@@ -127,7 +127,13 @@ impl MudHandler {
         let others: Vec<_> = state
             .players_in_room(&room_id, self.id)
             .iter()
-            .map(|c| (c.channel, c.handle.clone()))
+            .filter_map(|c| {
+                if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
+                    Some((ch, h.clone()))
+                } else {
+                    None
+                }
+            })
             .collect();
 
         let room_view = render_entry_room(&state, &room_id, &player_name, self.id);
@@ -170,8 +176,8 @@ impl MudHandler {
             self.username.clone(),
             race_id,
             class_id,
-            channel,
+            Some(channel),
-            handle,
+            Some(handle),
         );
         state.save_player_to_db(self.id);
         drop(state);
@@ -203,7 +209,13 @@ impl MudHandler {
             let others: Vec<_> = state
                 .players_in_room(&conn.player.room_id, self.id)
                 .iter()
-                .map(|c| (c.channel, c.handle.clone()))
+                .filter_map(|c| {
+                    if let (Some(ch), Some(h)) = (c.channel, &c.handle) {
+                        Some((ch, h.clone()))
+                    } else {
+                        None
+                    }
+                })
                 .collect();
             drop(state);
             for (ch, h) in others {
@@ -415,7 +427,7 @@ impl russh::server::Handler for MudHandler {
                     }
 
                     let keep_going =
-                        commands::execute(&line, self.id, &self.state, session, channel)
+                        commands::execute_for_ssh(&line, self.id, &self.state, session, channel)
                             .await?;
                     if !keep_going {
                         self.handle_disconnect().await;

src/tick.rs

@@ -291,11 +291,15 @@ pub async fn run_tick_engine(state: SharedState) {
                     return None;
                 }
                 let conn = st.players.get(&pid)?;
+                if let (Some(ch), Some(h)) = (conn.channel, &conn.handle) {
                     Some((
-                    conn.channel,
+                        ch,
-                    conn.handle.clone(),
+                        h.clone(),
                         format!("{}{}", msg, ansi::prompt()),
                     ))
+                } else {
+                    None
+                }
             })
             .collect();