Merge PR #3076: Issue #3069: --connect resends a command that crossed the app's idle-close notice

apps/lightcraft/src/control_server.rs

@@ -204,15 +204,40 @@ fn parse_line(bytes: &[u8]) -> Line {
     }
 }
 
+/// The error of the notice sent when an idle connection is closed. Clients match the reply's
+/// `"closed": "idle"` marker; `crates/mcp/src/backend.rs` also matches this text, for apps from
+/// before the marker, so keep the two in step.
+const IDLE_CLOSE_ERROR: &str = "control connection idle read timed out; closing the connection";
+
+/// Why the server closes a connection; its error reply says so.
+#[derive(Debug, PartialEq)]
+enum Closing {
+    /// Nothing complete arrived within the idle timeout. Nothing more is read, so a request that
+    /// crosses the notice is never run: the reply carries `"closed": "idle"` to tell clients it is
+    /// safe to send that request again on a new connection.
+    Idle,
+    /// Junk, an overlong line or a failed read.
+    Error(String),
+}
+
+impl Closing {
+    fn reply(&self) -> Value {
+        match self {
+            Self::Idle => json!({"ok": false, "error": IDLE_CLOSE_ERROR, "closed": "idle"}),
+            Self::Error(e) => json!({"ok": false, "error": e}),
+        }
+    }
+}
+
 /// Read one `\n`-terminated line of at most `max` bytes into `buf` (newline stripped).
 /// `Ok(false)` at end of stream; `Err` when the line is too long or the read fails.
-fn read_bounded_line(reader: &mut impl BufRead, buf: &mut Vec<u8>, max: usize) -> Result<bool, String> {
+fn read_bounded_line(reader: &mut impl BufRead, buf: &mut Vec<u8>, max: usize) -> Result<bool, Closing> {
     buf.clear();
     // `take` bounds the bytes buffered for one line: a peer can't make us allocate more than `max`.
     let limit = u64::try_from(max).unwrap_or(u64::MAX).saturating_add(1);
     let n = reader.take(limit).read_until(b'\n', buf).map_err(|e| match e.kind() {
-        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut => "control connection idle read timed out; closing the connection".to_string(),
-        _ => e.to_string(),
+        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut => Closing::Idle,
+        _ => Closing::Error(e.to_string()),
     })?;
     if n == 0 {
         return Ok(false);
@@ -225,7 +250,7 @@ fn read_bounded_line(reader: &mut impl BufRead, buf: &mut Vec<u8>, max: usize) -
         return Ok(true);
     }
     if buf.len() > max {
-        return Err(format!("request line longer than {max} bytes; closing the connection"));
+        return Err(Closing::Error(format!("request line longer than {max} bytes; closing the connection")));
     }
     // Last line without a newline, then end of stream.
     Ok(true)
@@ -238,12 +263,12 @@ fn serve_lines<R: Read, W: Write>(read: R, mut out: W, mut handle: impl FnMut(St
     let mut reader = BufReader::new(read);
     let mut buf = Vec::new();
     loop {
-        let reject = match read_bounded_line(&mut reader, &mut buf, MAX_LINE) {
+        let closing = match read_bounded_line(&mut reader, &mut buf, MAX_LINE) {
             Ok(false) => return,
-            Err(e) => e,
+            Err(closing) => closing,
             Ok(true) => match parse_line(&buf) {
                 Line::Blank => continue,
-                Line::Reject(e) => e,
+                Line::Reject(e) => Closing::Error(e),
                 Line::Request { id, method, params } => {
                     let Some(mut r) = handle(method, params) else { return };
                     if let Some(o) = r.as_object_mut() {
@@ -256,7 +281,7 @@ fn serve_lines<R: Read, W: Write>(read: R, mut out: W, mut handle: impl FnMut(St
                 }
             },
         };
-        let _ = writeln!(out, "{}", json!({"ok": false, "error": reject})).and_then(|()| out.flush());
+        let _ = writeln!(out, "{}", closing.reply()).and_then(|()| out.flush());
         return;
     }
 }
@@ -321,7 +346,9 @@ mod tests {
             assert!(connection.requests.try_recv().is_err(), "incomplete input never runs");
             let reply: Value = serde_json::from_str(&reply).unwrap();
             assert_eq!(reply["ok"], false);
-            assert!(reply["error"].as_str().unwrap().contains("idle read timed out"));
+            assert_eq!(reply["error"], IDLE_CLOSE_ERROR);
+            assert_eq!(reply["closed"], "idle", "clients may resend a request that crossed this notice");
+            assert!(reply.get("id").is_none());
 
             // A subsequent connection is admitted through the same accounting path.
             let mut next = Connection::new(&open, Duration::from_secs(1), |c| c.shutdown(std::net::Shutdown::Write).unwrap());
@@ -447,6 +474,7 @@ mod tests {
         assert_eq!(r.len(), 1);
         assert_eq!(r[0]["ok"], false);
         assert!(r[0]["error"].as_str().unwrap().contains("closing the connection"));
+        assert!(r[0].get("closed").is_none(), "only an idle close invites a resend: {}", r[0]);
     }
 
     #[test]

crates/mcp/src/backend.rs

@@ -136,39 +136,58 @@ fn peer_closed(stream: &TcpStream) -> bool {
     stream.set_nonblocking(false).is_err() || !open
 }
 
+/// The app's idle-close notice (`apps/lightcraft/src/control_server.rs`), matched when a reply lacks
+/// the `closed` marker (an app from before it). Keep the two in step.
+const IDLE_CLOSE_ERROR: &str = "control connection idle read timed out; closing the connection";
+
+/// True for the app's idle-close notice: an id-less error the app writes when it closes a connection
+/// on which it read nothing for its idle timeout. Read in reply to a request, it means the app closed
+/// the connection without reading the request (the request crossed the notice), so nothing ran.
+fn is_idle_close(reply: &Value) -> bool {
+    reply.get("id").is_none()
+        && reply.get("ok").and_then(Value::as_bool) == Some(false)
+        && (reply.get("closed").and_then(Value::as_str) == Some("idle") || reply.get("error").and_then(Value::as_str) == Some(IDLE_CLOSE_ERROR))
+}
+
 impl Backend for Remote {
     fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
         let id = self.next_id;
         self.next_id += 1;
         let line = json!({"id": id, "method": method, "params": params}).to_string();
-        // 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((_, false)) => {
-                self.conn = None;
-                self.roundtrip(&line).map_err(|(e, sent)| {
+        // Send again once, on a fresh connection, but only when the request never reached the app: it
+        // could not be written (the app restarted), or the app's idle-close notice crossed it (the app
+        // stopped reading before it arrived). After a full write anything else may mean the command
+        // ran, so resending it could apply a mutation twice (toggle, undo, ...): report the unknown
+        // outcome instead. One resend at most, whatever the reason, so this can't loop.
+        let mut resent = false;
+        let v = loop {
+            let reply = match self.roundtrip(&line) {
+                Ok(r) => r,
+                Err((e, sent)) => {
                     self.conn = None;
                     if sent {
-                        self.unknown_outcome(method, &e)
-                    } else {
-                        format!("LightCraft app at {} is not reachable: {e} (start it with `lightcraft --control PORT`)", self.addr)
+                        return Err(self.unknown_outcome(method, &e));
                     }
-                })?
-            }
-            Err((e, true)) => {
+                    if resent {
+                        return Err(format!("LightCraft app at {} is not reachable: {e} (start it with `lightcraft --control PORT`)", self.addr));
+                    }
+                    resent = true;
+                    continue;
+                }
+            };
+            let v = parse_reply(&reply, id).map_err(|e| {
                 self.conn = None;
-                return Err(self.unknown_outcome(method, &e));
+                self.unknown_outcome(method, &e)
+            })?;
+            if v.get("id").is_none() {
+                self.conn = None;
+                if !resent && is_idle_close(&v) {
+                    resent = true;
+                    continue;
+                }
             }
+            break v;
         };
-        let v = parse_reply(&reply, id).map_err(|e| {
-            self.conn = None;
-            self.unknown_outcome(method, &e)
-        })?;
-        if v.get("id").is_none() {
-            self.conn = None;
-        }
         if v.get("ok").and_then(Value::as_bool) == Some(true) {
             Ok(v.get("result").cloned().unwrap_or(Value::Null))
         } else {
@@ -275,6 +294,11 @@ mod tests {
             ("{\"id\":1,\"ok\":true,\"result\":null}\r\n", Ok(Value::Null)),
             ("{\"id\":1,\"ok\":false,\"error\":\"unknown command\"}\n", Err("unknown command".to_string())),
             ("{\"ok\":false,\"error\":\"too many control connections\"}\n", Err("too many control connections".to_string())),
+            // The idle-close text in reply to the request (it has the id) is the command's own error.
+            (
+                "{\"id\":1,\"ok\":false,\"error\":\"control connection idle read timed out; closing the connection\"}\n",
+                Err(IDLE_CLOSE_ERROR.to_string()),
+            ),
         ] {
             with_reply(reply.as_bytes(), |remote, result| {
                 assert_eq!(result, expected);
@@ -380,51 +404,148 @@ mod tests {
         assert_eq!(seen.load(Ordering::SeqCst), 2);
     }
 
+    /// Accept the next connection within 5 s, as a blocking stream with a 3 s read timeout.
+    fn accept(listener: &TcpListener) -> (BufReader<TcpStream>, TcpStream) {
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        let stream = loop {
+            match listener.accept() {
+                Ok((stream, _)) => break stream,
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock && std::time::Instant::now() < deadline => {
+                    std::thread::sleep(Duration::from_millis(5));
+                }
+                Err(e) => panic!("accept: {e}"),
+            }
+        };
+        // BSD kernels (macOS, FreeBSD) let an accepted socket inherit the listener's non-blocking flag.
+        stream.set_nonblocking(false).unwrap();
+        stream.set_read_timeout(Some(Duration::from_secs(3))).unwrap();
+        (BufReader::new(stream.try_clone().unwrap()), stream)
+    }
+
+    fn read_request(reader: &mut BufReader<TcpStream>) -> Value {
+        let mut line = String::new();
+        reader.read_line(&mut line).unwrap();
+        serde_json::from_str(&line).unwrap()
+    }
+
+    /// The app's idle-close notice, as `apps/lightcraft/src/control_server.rs` writes it.
+    fn idle_notice() -> Value {
+        json!({"ok": false, "error": IDLE_CLOSE_ERROR, "closed": "idle"})
+    }
+
+    /// Issue #3069. The app closes an idle connection with an id-less notice. Either the notice
+    /// arrives before the next request (the client sees the close and reconnects before sending),
+    /// or the request crosses it (sent just as the app stopped reading): the notice then comes
+    /// back as the request's reply although the app never ran it, so the client sends it once more
+    /// on a new connection. Both orders are forced here, so the test does not depend on timing.
     #[test]
     fn idle_close_reconnects_without_replaying_a_command() {
+        let _serial = crate::serial_sockets_and_children();
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();
         listener.set_nonblocking(true).unwrap();
         let addr = listener.local_addr().unwrap().to_string();
         let (closed, expired) = std::sync::mpsc::channel();
         let server = std::thread::spawn(move || {
-            let mut ids = Vec::new();
-            for _ in 0..2 {
-                let deadline = std::time::Instant::now() + Duration::from_secs(5);
-                let stream = loop {
-                    match listener.accept() {
-                        Ok((stream, _)) => break stream,
-                        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock && std::time::Instant::now() < deadline => {
-                            std::thread::sleep(Duration::from_millis(5));
-                        }
-                        Err(e) => panic!("accept: {e}"),
-                    }
-                };
-                stream.set_read_timeout(Some(Duration::from_secs(3))).unwrap();
-                let mut reader = BufReader::new(stream.try_clone().unwrap());
-                let mut writer = stream;
-                let mut line = String::new();
-                reader.read_line(&mut line).unwrap();
-                let request: Value = serde_json::from_str(&line).unwrap();
-                ids.push(request["id"].as_u64().unwrap());
-                writeln!(writer, "{}", json!({"id": request["id"], "ok": true, "result": ids.len()})).unwrap();
-                writer.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
-                line.clear();
-                let error = reader.read_line(&mut line).unwrap_err();
-                assert!(matches!(error.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut));
-                assert!(line.is_empty(), "no command is resent while idle");
-                writeln!(writer, "{}", json!({"ok": false, "error": "control connection idle read timed out; closing the connection"})).unwrap();
-                drop(reader);
-                drop(writer);
-                closed.send(()).unwrap();
-            }
-            ids
+            let mut ran = Vec::new();
+            let mut run = |request: &Value, writer: &mut TcpStream| {
+                ran.push(request["id"].as_u64().unwrap());
+                writeln!(writer, "{}", json!({"id": request["id"], "ok": true, "result": ran.len()})).unwrap();
+            };
+
+            // 1. Idle with nothing sent: the notice (in an older app's form, without `closed`) and
+            //    the close reach the client before its next request.
+            let (mut reader, mut writer) = accept(&listener);
+            run(&read_request(&mut reader), &mut writer);
+            writer.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
+            let mut line = String::new();
+            let error = reader.read_line(&mut line).unwrap_err();
+            assert!(matches!(error.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut));
+            assert!(line.is_empty(), "no command is resent while idle");
+            writeln!(writer, "{}", json!({"ok": false, "error": IDLE_CLOSE_ERROR})).unwrap();
+            drop((reader, writer));
+            closed.send(()).unwrap();
+
+            // 2. The client reconnected before sending. Its next request crosses the notice: the
+            //    app's read timed out just before the request arrived, so it is not run. (Reading
+            //    it here only shows that it was sent on this connection and keeps the close clean.)
+            let (mut reader, mut writer) = accept(&listener);
+            run(&read_request(&mut reader), &mut writer);
+            let crossed = read_request(&mut reader);
+            writeln!(writer, "{}", idle_notice()).unwrap();
+            drop((reader, writer));
+
+            // 3. The crossed request, sent once more on a new connection, and nothing after it.
+            let (mut reader, mut writer) = accept(&listener);
+            let resent = read_request(&mut reader);
+            assert_eq!(resent, crossed, "the same request is sent again");
+            run(&resent, &mut writer);
+            line.clear();
+            assert_eq!(reader.read_line(&mut line).unwrap(), 0, "nothing else arrives: {line}");
+            ran
         });
         let mut remote = Remote::lazy(&addr);
         assert_eq!(remote.call("engine.execute", json!({"command": "edit.undo"})).unwrap(), json!(1));
         expired.recv_timeout(Duration::from_secs(5)).unwrap();
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        while !remote.conn.as_ref().is_some_and(|(_, writer)| peer_closed(writer)) {
+            assert!(std::time::Instant::now() < deadline, "the idle close never reached the client");
+            std::thread::sleep(Duration::from_millis(1));
+        }
         assert_eq!(remote.call("engine.execute", json!({"command": "edit.undo"})).unwrap(), json!(2));
-        expired.recv_timeout(Duration::from_secs(5)).unwrap();
-        assert_eq!(server.join().unwrap(), [1, 2], "each command ran once on a separate connection");
+        assert_eq!(remote.call("engine.execute", json!({"command": "edit.undo"})).unwrap(), json!(3));
+        drop(remote);
+        assert_eq!(server.join().unwrap(), [1, 2, 3], "each command ran once");
+    }
+
+    /// The resend after an idle close happens once: a second notice is the call's error.
+    #[test]
+    fn idle_close_is_resent_only_once() {
+        let _serial = crate::serial_sockets_and_children();
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        listener.set_nonblocking(true).unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        let server = std::thread::spawn(move || {
+            for _ in 0..2 {
+                let (mut reader, mut writer) = accept(&listener);
+                assert_eq!(read_request(&mut reader)["id"], 1);
+                writeln!(writer, "{}", idle_notice()).unwrap();
+            }
+            listener
+        });
+        let mut remote = Remote::lazy(&addr);
+        assert_eq!(remote.call("engine.execute", json!({"command": "edit.undo"})), Err(IDLE_CLOSE_ERROR.to_string()));
+        assert!(remote.conn.is_none(), "the closed connection is not reused");
+        drop(remote);
+        let listener = server.join().unwrap();
+        assert_eq!(listener.accept().unwrap_err().kind(), std::io::ErrorKind::WouldBlock, "a third connection was opened");
+    }
+
+    #[test]
+    fn only_the_idle_close_notice_counts_as_undelivered() {
+        let idle = [
+            json!({"ok": false, "error": IDLE_CLOSE_ERROR, "closed": "idle"}),
+            // an app from before the `closed` marker
+            json!({"ok": false, "error": IDLE_CLOSE_ERROR}),
+            // the marker decides when present, whatever the wording
+            json!({"ok": false, "error": "idle, closing", "closed": "idle"}),
+        ];
+        for reply in idle {
+            assert!(is_idle_close(&reply), "{reply}");
+        }
+        let delivered = [
+            // a reply to the request itself
+            json!({"id": 1, "ok": false, "error": IDLE_CLOSE_ERROR}),
+            json!({"id": 1, "ok": false, "error": IDLE_CLOSE_ERROR, "closed": "idle"}),
+            // other refusals and closes
+            json!({"ok": false, "error": "too many control connections"}),
+            json!({"ok": false, "error": "bad JSON; closing the connection"}),
+            json!({"ok": false, "error": "bad JSON; closing the connection", "closed": "junk"}),
+            json!({"ok": false, "error": format!("{IDLE_CLOSE_ERROR}.")}),
+            json!({"ok": true, "result": IDLE_CLOSE_ERROR, "closed": "idle"}),
+        ];
+        for reply in delivered {
+            assert!(!is_idle_close(&reply), "{reply}");
+        }
     }
 
     #[test]

docs/control-protocol.md

@@ -21,15 +21,25 @@ headless snapshots or `lightcraft-cli mcp` without `--connect`. A request that h
 
 Connections close after five minutes without incoming bytes while the server is waiting for
 input, including in the middle of an unfinished request line. This allows pauses between
-automation commands while releasing abandoned connection slots. An idle read timeout gets an
-`ok: false` error saying the connection is closing; incomplete input is never dispatched.
+automation commands while releasing abandoned connection slots. An idle read timeout gets one
+reply without an `id`, then the connection closes; incomplete input is never dispatched:
+
+```text
+← {"ok": false, "error": "control connection idle read timed out; closing the connection", "closed": "idle"}
+```
+
 This is a socket read inactivity limit, not a total connection lifetime: a peer that keeps
 sending bytes can keep it open. Waiting for an already-dispatched command's reply does not
 count as idle reading and retains the request timeout above.
 
-Reconnect before sending another request after an idle close. MCP connect mode detects closed
-connections before sending and reconnects; if a connection is lost after sending a request,
-its outcome remains unknown and the client must not blindly replay it.
+Reconnect before sending another request after an idle close. A request sent just as the
+connection expires crosses the notice: the client reads the notice where it expects the reply.
+The server reads nothing after its idle timeout, so that request never ran, and `"closed": "idle"`
+(only on this notice) says it is safe to send it again, once, on a new connection. Apps from
+before the marker send the same notice without it; match its exact `error` text for them. MCP
+connect mode does both: it detects closed connections before sending and reconnects, and
+resends a request that crossed the idle notice once. Any other lost connection after sending a
+request leaves its outcome unknown, and the client must not blindly replay it.
 
 **Only requests are read.** Every line must be a JSON object with a string `method` (`id` and `params` are
 optional; blank lines are skipped). Anything else — text that isn't JSON, a JSON array or number, an object without

docs/mcp.md

@@ -22,7 +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
+A request is sent again (once) only when it never reached the app: it could not be sent, or the app's
+idle-close notice came back in its place (the app closed the idle connection without reading it; see
+[control-protocol.md](control-protocol.md)). 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. Replies must be
 newline-terminated, at most 64 MiB before the newline (including encoded previews/screenshots),