MCP: do not resend a request whose reply was lost, report an unknown outcome (#1526)

crates/mcp/src/backend.rs

@@ -65,22 +65,46 @@ impl Remote {
         Err(last)
     }
 
-    fn roundtrip(&mut self, line: &str) -> std::io::Result<String> {
+    /// Send one request line and read the reply line. On failure the flag says whether the request
+    /// may have reached the app (written in full); only an unsent request is safe to resend.
+    fn roundtrip(&mut self, line: &str) -> Result<String, (std::io::Error, bool)> {
+        // A connection the app already closed (it restarted, or closed it after an error reply) is
+        // replaced before anything is sent on it, so that case still reconnects without a resend.
+        if self.conn.as_ref().is_some_and(|(reader, writer)| !reader.buffer().is_empty() || peer_closed(writer)) {
+            self.conn = None;
+        }
         if self.conn.is_none() {
-            self.reconnect()?;
+            self.reconnect().map_err(|e| (e, false))?;
         }
         let Some((reader, writer)) = self.conn.as_mut() else {
-            return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected"));
+            return Err((std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected"), false));
         };
-        writer.write_all(line.as_bytes())?;
-        writer.write_all(b"\n")?;
-        writer.flush()?;
+        let mut data = Vec::with_capacity(line.len() + 1);
+        data.extend_from_slice(line.as_bytes());
+        data.push(b'\n');
+        writer.write_all(&data).map_err(|e| (e, false))?;
+        writer.flush().map_err(|e| (e, true))?;
         let mut reply = String::new();
-        if reader.read_line(&mut reply)? == 0 {
-            return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed"));
+        if reader.read_line(&mut reply).map_err(|e| (e, true))? == 0 {
+            return Err((std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed"), true));
         }
         Ok(reply)
     }
+
+    fn unknown_outcome(&self, method: &str, e: &std::io::Error) -> String {
+        let addr = &self.addr;
+        format!("lost the LightCraft app at {addr} after sending `{method}` ({e}); the outcome is unknown and it was not resent, check the app")
+    }
+}
+
+/// True when the app has closed `stream` (or it is unusable), checked without blocking or consuming data.
+fn peer_closed(stream: &TcpStream) -> bool {
+    if stream.set_nonblocking(true).is_err() {
+        return true;
+    }
+    let mut byte = [0u8; 1];
+    let open = matches!(stream.peek(&mut byte), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock);
+    stream.set_nonblocking(false).is_err() || !open
 }
 
 impl Backend for Remote {
@@ -88,16 +112,26 @@ impl Backend for Remote {
         let id = self.next_id;
         self.next_id += 1;
         let line = json!({"id": id, "method": method, "params": params}).to_string();
-        // One retry with a fresh connection (the app may have restarted).
+        // Retry once with a fresh connection (the app may have restarted), but only when the request
+        // never reached the app. After a full write the command may have run, so resending it could
+        // apply a mutation twice (toggle, undo, ...): report the unknown outcome instead.
         let reply = match self.roundtrip(&line) {
             Ok(r) => r,
-            Err(_) => {
+            Err((_, false)) => {
                 self.conn = None;
-                self.roundtrip(&line).map_err(|e| {
+                self.roundtrip(&line).map_err(|(e, sent)| {
                     self.conn = None;
-                    format!("LightCraft app at {} is not reachable: {e} (start it with `lightcraft --control PORT`)", self.addr)
+                    if sent {
+                        self.unknown_outcome(method, &e)
+                    } else {
+                        format!("LightCraft app at {} is not reachable: {e} (start it with `lightcraft --control PORT`)", self.addr)
+                    }
                 })?
             }
+            Err((e, true)) => {
+                self.conn = None;
+                return Err(self.unknown_outcome(method, &e));
+            }
         };
         let v: Value = serde_json::from_str(reply.trim()).map_err(|e| format!("bad reply from app: {e}"))?;
         if v.get("ok").and_then(Value::as_bool) == Some(true) {
@@ -115,3 +149,66 @@ impl Backend for Remote {
         format!("remote {}", self.addr)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::TcpListener;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[test]
+    fn lost_reply_is_not_resent() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        let seen = Arc::new(AtomicUsize::new(0));
+        let seen2 = seen.clone();
+        std::thread::spawn(move || {
+            // Every connection: read the request, count it, drop without replying.
+            while let Ok((s, _)) = listener.accept() {
+                let mut line = String::new();
+                if BufReader::new(s).read_line(&mut line).unwrap_or(0) > 0 {
+                    seen2.fetch_add(1, Ordering::SeqCst);
+                }
+            }
+        });
+        let mut remote = Remote::lazy(&addr);
+        let err = remote.call("engine.execute", json!({})).unwrap_err();
+        assert!(err.contains("outcome is unknown"), "{err}");
+        std::thread::sleep(Duration::from_millis(200));
+        assert_eq!(seen.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn connection_closed_by_app_reconnects_before_sending() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        let seen = Arc::new(AtomicUsize::new(0));
+        let seen2 = seen.clone();
+        std::thread::spawn(move || {
+            // Every connection: answer one request, then close (as after an app restart).
+            while let Ok((s, _)) = listener.accept() {
+                let mut writer = s.try_clone().unwrap();
+                let mut line = String::new();
+                if BufReader::new(s).read_line(&mut line).unwrap_or(0) > 0 {
+                    seen2.fetch_add(1, Ordering::SeqCst);
+                    writer.write_all(b"{\"ok\": true, \"result\": 7}\n").unwrap();
+                }
+            }
+        });
+        let mut remote = Remote::lazy(&addr);
+        assert_eq!(remote.call("engine.commands", json!({})).unwrap(), json!(7));
+        std::thread::sleep(Duration::from_millis(200));
+        assert_eq!(remote.call("engine.commands", json!({})).unwrap(), json!(7));
+        assert_eq!(seen.load(Ordering::SeqCst), 2);
+    }
+
+    #[test]
+    fn unreachable_app_is_reported() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        drop(listener);
+        let err = Remote::lazy(&addr).call("engine.commands", json!({})).unwrap_err();
+        assert!(err.contains("not reachable"), "{err}");
+    }
+}

docs/mcp.md

@@ -22,6 +22,9 @@ it open, `--library` on the same folder fails with "This library is already open
 `--demo` starts the headless session with the procedurally generated demo library;
 `--compact` lists only the helper tools (see below). In connect mode the server starts even when
 the app is not running yet and connects on the first call (and reconnects if the app restarts).
+A request is sent again only when it never reached the app. If the connection drops after a request
+was sent but before its reply, the tool fails with "the outcome is unknown" and nothing is resent, so a
+command such as undo can't run twice: check the app's state before retrying.
 
 Logs go to stderr; stdout carries only protocol messages.