diff --git a/crates/mcp/src/backend.rs b/crates/mcp/src/backend.rs index 90bedd94..52226eb7 100644 --- a/crates/mcp/src/backend.rs +++ b/crates/mcp/src/backend.rs @@ -42,6 +42,11 @@ impl Remote { &self.addr } + #[cfg(test)] + pub(crate) fn connection_probe(&self) -> TcpStream { + self.conn.as_ref().unwrap().1.try_clone().unwrap() + } + fn reconnect(&mut self) -> std::io::Result<()> { self.conn = None; let mut last = std::io::Error::new(std::io::ErrorKind::NotFound, format!("cannot resolve {}", self.addr)); @@ -61,10 +66,26 @@ impl Remote { Err(last) } - fn roundtrip(&mut self, line: &str) -> std::io::Result { + fn prepare_connection(&mut self) -> std::io::Result<()> { + if let Some((_, stream)) = self.conn.as_ref() { + stream.set_nonblocking(true)?; + let probe = stream.peek(&mut [0]); + stream.set_nonblocking(false)?; + match probe { + // Closed by the app, or holding bytes nobody asked for (an app that explains why it + // closes): either way no reply to the next request can be read from it. + Ok(_) => self.conn = None, + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} + Err(_) => self.conn = None, + } + } if self.conn.is_none() { self.reconnect()?; } + Ok(()) + } + + fn roundtrip(&mut self, line: &str) -> std::io::Result { let Some((reader, writer)) = self.conn.as_mut() else { return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected to the app")); }; @@ -86,6 +107,10 @@ impl Remote { impl Backend for Remote { fn call(&mut self, method: &str, params: Value) -> Result { + self.prepare_connection().map_err(|e| { + self.conn = None; + format!("VectorCraft app at {} is not reachable: {e}; `{method}`: nothing was sent", self.addr) + })?; let id = self.next_id; self.next_id += 1; let line = json!({"id": id, "method": method, "params": params}).to_string(); diff --git a/crates/mcp/src/tests.rs b/crates/mcp/src/tests.rs index 603b486c..0f08e89c 100644 --- a/crates/mcp/src/tests.rs +++ b/crates/mcp/src/tests.rs @@ -586,6 +586,140 @@ fn remote_connect_fails_fast() { drop(held); } +fn reusable_listener(addr: std::net::SocketAddr) -> std::net::TcpListener { + let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(); + socket.set_reuse_address(true).unwrap(); + socket.bind(&addr.into()).unwrap(); + socket.listen(1).unwrap(); + socket.into() +} + +fn wait_for_remote_eof(remote: &Remote) { + let probe = remote.connection_probe(); + probe.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap(); + assert_eq!(probe.peek(&mut [0]).unwrap(), 0); +} + +#[test] +fn remote_edit_succeeds_once_when_app_restarts() { + let listener = reusable_listener(std::net::SocketAddr::from(([127, 0, 0, 1], 0))); + let addr = listener.local_addr().unwrap(); + let mut remote = Remote::connect(&addr.to_string()).unwrap(); + let (old_connection, _) = listener.accept().unwrap(); + let old_handle = std::thread::spawn(move || { + let timeout = std::time::Duration::from_secs(2); + old_connection.set_read_timeout(Some(timeout)).unwrap(); + old_connection.set_write_timeout(Some(timeout)).unwrap(); + let mut reader = BufReader::new(old_connection.try_clone().unwrap()); + let mut writer = old_connection; + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + let message: Value = serde_json::from_str(&line).unwrap(); + assert_eq!(message["method"], "document.inspect"); + let result = Headless::with_document().call("document.inspect", message["params"].clone()).unwrap(); + writeln!(writer, "{}", json!({"id": message["id"], "ok": true, "result": result})).unwrap(); + writer + }); + let inspected = remote.call("document.inspect", json!({})).unwrap(); + assert!(inspected["objects"].as_u64().unwrap() > 0); + drop(old_handle.join().unwrap()); + drop(listener); + wait_for_remote_eof(&remote); + + let listener = reusable_listener(addr); + listener.set_nonblocking(true).unwrap(); + let handle = std::thread::spawn(move || { + let timeout = std::time::Duration::from_secs(2); + let deadline = std::time::Instant::now() + timeout; + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + if std::time::Instant::now() >= deadline { + return (0, 0); + } + std::thread::yield_now(); + } + Err(error) => panic!("accept failed: {error}"), + } + }; + stream.set_nonblocking(false).unwrap(); + stream.set_read_timeout(Some(timeout)).unwrap(); + stream.set_write_timeout(Some(timeout)).unwrap(); + let mut writer = stream.try_clone().unwrap(); + let mut backend = Headless::with_document(); + let base = backend.call("document.inspect", json!({})).unwrap()["objects"].as_u64().unwrap(); + let mut requests = 0; + for line in BufReader::new(stream).lines() { + let message: Value = serde_json::from_str(&line.unwrap()).unwrap(); + let result = backend.call(message["method"].as_str().unwrap(), message["params"].clone()).unwrap(); + requests += 1; + writeln!(writer, "{}", json!({"id": message["id"], "ok": true, "result": result})).unwrap(); + } + let objects = backend.call("document.inspect", json!({})).unwrap()["objects"].as_u64().unwrap(); + (requests, objects - base) + }); + + let edit = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {"x": 10, "y": 10, "width": 20, "height": 20}})); + drop(remote); + let (requests, edits) = handle.join().unwrap(); + + assert!(edit.is_ok(), "the first edit after restart must succeed: {edit:?}"); + assert_eq!((requests, edits), (1, 1), "the edit must be sent and applied exactly once"); +} + +/// A connection holding a line nobody asked for (an app that says why it closes an idle +/// connection) is not used: the call goes out once, on a fresh connection, and succeeds. +#[test] +fn remote_does_not_send_on_a_connection_with_a_stray_line() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let mut remote = Remote::connect(&addr).unwrap(); + let (mut stale, _) = listener.accept().unwrap(); + writeln!(stale, "{}", json!({"ok": false, "error": "idle"})).unwrap(); + let probe = remote.connection_probe(); + probe.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap(); + assert_eq!(probe.peek(&mut [0]).unwrap(), 1, "the stray line arrived"); + drop(probe); + let h = std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap(); + let mut out = stream.try_clone().unwrap(); + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).unwrap(); + let msg: Value = serde_json::from_str(&line).unwrap(); + writeln!(out, "{}", json!({"ok": true, "result": 7, "id": msg["id"]})).unwrap(); + }); + assert_eq!(remote.call("document.inspect", json!({})), Ok(json!(7))); + h.join().unwrap(); + // Nothing was written to the stale connection: it ends (or is reset, its line unread) without a byte. + stale.set_read_timeout(Some(std::time::Duration::from_secs(2))).unwrap(); + let mut rest = String::new(); + assert_eq!(BufReader::new(stale).read_line(&mut rest).unwrap_or(0), 0, "{rest}"); +} + +#[test] +fn remote_call_reports_nothing_sent_when_reconnect_is_refused() { + let listener = reusable_listener(std::net::SocketAddr::from(([127, 0, 0, 1], 0))); + let addr = listener.local_addr().unwrap(); + let mut remote = Remote::connect(&addr.to_string()).unwrap(); + let (old_connection, _) = listener.accept().unwrap(); + drop(old_connection); + drop(listener); + wait_for_remote_eof(&remote); + let held = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(); + held.set_reuse_address(true).unwrap(); + held.bind(&addr.into()).unwrap(); + + let start = std::time::Instant::now(); + let error = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {"width": 20, "height": 20}})).unwrap_err(); + + assert!(start.elapsed() < std::time::Duration::from_secs(2)); + assert!(error.contains("not reachable"), "{error}"); + assert!(error.contains("nothing was sent"), "{error}"); + assert!(!error.contains("operation may have completed"), "{error}"); +} + #[test] fn headless_backend_methods() { let mut h = Headless::new(); diff --git a/docs/mcp.md b/docs/mcp.md index 958f88f5..0ef5d5ce 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -111,7 +111,11 @@ to) before it is dispatched. When a tool call reaches a running app, it is one c request, and the app echoes its `id` in the reply: a reply with any other `id` is rejected, the connection dropped, and the next call starts a fresh one — the call reports uncertainty like any reply that never arrived whole. The app's refusals carry no `id` (a connection past its 16, a line it -would not read): the call reports the app's reason and says nothing was run. Clients requesting `2026-07-28` through `initialize`, or through a request's +would not read): the call reports the app's reason and says nothing was run. Before each call the +server checks that the app still holds the connection and opens a new one when it doesn't (the app +closes a connection left idle for 5 minutes, and a restarted app is a new one), so a request is +never written to a closed socket; when the app can't be reached, the call says that nothing was +sent. Clients requesting `2026-07-28` through `initialize`, or through a request's `_meta["io.modelcontextprotocol/protocolVersion"]`, receive `resultType: "complete"`, `ttlMs` and `cacheScope: "private"` on tool/resource/template/prompt lists and resource/prompt reads. List TTL is 600000 ms; live reads use zero. Older clients keep the original response shapes.