Merge upstream/main into PR #3195

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/wordcraft/src/control_server.rs

@@ -9,6 +9,7 @@ use std::io::{BufRead, BufReader, Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc::{Receiver, Sender, channel};
 use std::time::{Duration, Instant};
 
@@ -21,6 +22,38 @@ const MAX_LINE: usize = 1 << 20;
 /// A connection that sends no authorised request in this time is closed.
 const FIRST_REQUEST: Duration = Duration::from_secs(30);
 
+/// How long a caller waits for the UI to answer a request. A request still queued after this
+/// is cancelled and never runs.
+const REPLY_WAIT: Duration = Duration::from_secs(60);
+
+/// Connections that have not yet sent an authorised request are limited to this many; more are
+/// closed at once. Each one holds a thread until its first request or [`FIRST_REQUEST`].
+const MAX_UNAUTHENTICATED: usize = 64;
+
+/// Counts one unauthenticated connection until it authenticates or ends.
+struct Admission(Option<Arc<AtomicUsize>>);
+
+impl Admission {
+    fn enter(count: &Arc<AtomicUsize>, max: usize) -> Option<Self> {
+        if count.fetch_add(1, Ordering::SeqCst) >= max {
+            count.fetch_sub(1, Ordering::SeqCst);
+            return None;
+        }
+        Some(Self(Some(count.clone())))
+    }
+    fn release(&mut self) {
+        if let Some(c) = self.0.take() {
+            c.fetch_sub(1, Ordering::SeqCst);
+        }
+    }
+}
+
+impl Drop for Admission {
+    fn drop(&mut self) {
+        self.release();
+    }
+}
+
 /// The key file of a running control server; [`KeyFile::remove`] it when the window closes.
 pub struct KeyFile(PathBuf);
 
@@ -37,6 +70,15 @@ impl KeyFile {
 /// cannot be bound or there is no random key (the control channel is off); the key file is
 /// `None` when it could not be written (requests are then refused).
 pub fn start(port: u16, ctx: egui::Context, settings: Option<&Path>) -> Option<(Receiver<ControlRequest>, Option<KeyFile>)> {
+    start_with(port, ctx, settings, |listener, ctx, key| serve_on(listener, ctx, key, FIRST_REQUEST))
+}
+
+fn start_with(
+    port: u16,
+    ctx: egui::Context,
+    settings: Option<&Path>,
+    serve: impl FnOnce(TcpListener, egui::Context, String) -> std::io::Result<Receiver<ControlRequest>>,
+) -> Option<(Receiver<ControlRequest>, Option<KeyFile>)> {
     let listener = match TcpListener::bind(("127.0.0.1", port)) {
         Ok(l) => l,
         Err(e) => {
@@ -67,11 +109,21 @@ pub fn start(port: u16, ctx: egui::Context, settings: Option<&Path>) -> Option<(
             None
         }
     };
+    let rx = match serve(listener, ctx, key) {
+        Ok(rx) => rx,
+        Err(e) => {
+            if let Some(f) = &key_file {
+                f.remove();
+            }
+            log::error!("control server off: cannot start the accept thread ({e})");
+            return None;
+        }
+    };
     match &key_file {
         Some(f) => log::info!("control server listening on 127.0.0.1:{port}, key in {}", f.path().display()),
         None => log::info!("control server listening on 127.0.0.1:{port}"),
     }
-    Some((serve_on(listener, ctx, key, FIRST_REQUEST), key_file))
+    Some((rx, key_file))
 }
 
 /// A new key: 32 random bytes from the operating system, as hex.
@@ -103,16 +155,47 @@ fn keys_match(given: &str, key: &str) -> bool {
     !b.is_empty() && a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
 }
 
-fn serve_on(listener: TcpListener, ctx: egui::Context, key: String, first: Duration) -> Receiver<ControlRequest> {
+fn serve_on(listener: TcpListener, ctx: egui::Context, key: String, first: Duration) -> std::io::Result<Receiver<ControlRequest>> {
+    serve_with(listener, ctx, key, Limits { first, wait: REPLY_WAIT, max_unauthenticated: MAX_UNAUTHENTICATED })
+}
+
+/// The time and count budgets of a server (small ones in tests).
+#[derive(Clone, Copy)]
+struct Limits {
+    first: Duration,
+    wait: Duration,
+    max_unauthenticated: usize,
+}
+
+fn serve_with(listener: TcpListener, ctx: egui::Context, key: String, limits: Limits) -> std::io::Result<Receiver<ControlRequest>> {
+    serve_with_spawner(listener, ctx, key, limits, |accept| std::thread::Builder::new().name("control-accept".into()).spawn(accept).map(|_| ()))
+}
+
+fn serve_with_spawner(
+    listener: TcpListener,
+    ctx: egui::Context,
+    key: String,
+    limits: Limits,
+    spawn: impl FnOnce(Box<dyn FnOnce() + Send>) -> std::io::Result<()>,
+) -> std::io::Result<Receiver<ControlRequest>> {
     let (tx, rx) = channel::<ControlRequest>();
     let key: Arc<str> = key.into();
-    std::thread::spawn(move || {
+    let unauthenticated = Arc::new(AtomicUsize::new(0));
+    let accept = move || {
         for stream in listener.incoming().flatten() {
+            let Some(admission) = Admission::enter(&unauthenticated, limits.max_unauthenticated) else {
+                log::warn!("control server: too many unauthenticated connections, one closed");
+                continue;
+            };
             let (tx, ctx, key) = (tx.clone(), ctx.clone(), key.clone());
-            std::thread::spawn(move || serve(stream, tx, ctx, &key, first));
+            // A thread that cannot start closes this connection only (the guard releases its slot).
+            if let Err(e) = std::thread::Builder::new().name("control-conn".into()).spawn(move || serve(stream, tx, ctx, &key, limits, admission)) {
+                log::warn!("control server: no thread for a connection: {e}");
+            }
         }
-    });
-    rx
+    };
+    spawn(Box::new(accept))?;
+    Ok(rx)
 }
 
 /// The read side of a connection with an optional overall deadline: each read waits only for
@@ -175,11 +258,11 @@ fn discard_rest_of_line(r: &mut BufReader<Timed>) {
     }
 }
 
-fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context, key: &str, first: Duration) {
+fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context, key: &str, limits: Limits, mut admission: Admission) {
     let Ok(read) = stream.try_clone() else { return };
     let mut out = stream;
     // One overall deadline from accept to the first authorised request; after it, none.
-    let mut reader = BufReader::new(Timed { s: read, deadline: Some(Instant::now() + first) });
+    let mut reader = BufReader::new(Timed { s: read, deadline: Some(Instant::now() + limits.first) });
     let mut authorised = false;
     // The key an `auth` request gave this connection.
     let mut conn_key: Option<String> = None;
@@ -228,6 +311,7 @@ fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context, key:
         }
         if !authorised {
             authorised = true;
+            admission.release();
             reader.get_mut().set_deadline(None);
         }
         if method == "auth" {
@@ -237,12 +321,18 @@ fn serve(stream: TcpStream, tx: Sender<ControlRequest>, ctx: egui::Context, key:
             }
             continue;
         }
-        let (req, rrx) = ControlRequest::new(method, params);
+        let (req, rrx, waiter) = ControlRequest::with_waiter(method, params);
         if tx.send(req).is_err() {
             break;
         }
         ctx.request_repaint();
-        let r = rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"}));
+        let r = rrx.recv_timeout(limits.wait).unwrap_or_else(|_| {
+            if waiter.cancel() {
+                json!({"ok": false, "error": "timeout before execution: the request was not run and never will be"})
+            } else {
+                json!({"ok": false, "error": "timeout: outcome unknown, the request had started when the wait ended; inspect the document to check"})
+            }
+        });
         if !reply(r) {
             break;
         }
@@ -260,7 +350,7 @@ mod tests {
     fn boot_with(first: Duration) -> (u16, Receiver<ControlRequest>) {
         let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
         let port = listener.local_addr().unwrap().port();
-        (port, serve_on(listener, egui::Context::default(), KEY.to_string(), first))
+        (port, serve_on(listener, egui::Context::default(), KEY.to_string(), first).unwrap())
     }
 
     fn boot() -> (u16, Receiver<ControlRequest>) {
@@ -512,6 +602,40 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    fn an_accept_thread_spawn_failure_aborts_start_and_removes_the_key() {
+        let dir = temp_dir("spawn-failure");
+        let path = wordcraft_control_key::key_path(&dir, 0);
+        let mut bound = None;
+        let result = start_with(0, egui::Context::default(), Some(&dir), |listener, ctx, key| {
+            bound = Some(listener.local_addr().unwrap());
+            assert_eq!(std::fs::read_to_string(&path).unwrap(), key, "startup wrote the key before spawning");
+            serve_with_spawner(listener, ctx, key, small(150, 8), |accept| {
+                drop(accept);
+                Err(std::io::Error::other("injected accept thread spawn failure"))
+            })
+        });
+        assert!(result.is_none(), "a server without its accept thread must not report success");
+        assert!(!path.exists(), "failed startup must remove its published key");
+        let _listener = TcpListener::bind(bound.unwrap()).unwrap();
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    #[test]
+    fn a_complete_json_request_at_eof_can_reach_the_ui() {
+        // This is why a bridge write failure before the newline is outcome-unknown.
+        let (port, rx) = boot();
+        let mut c = Conn::open(port);
+        let body = json!({"id": 1, "key": KEY, "method": "engine.execute", "params": {"command": "text.insert", "params": {"text": "once"}}});
+        c.1.write_all(body.to_string().as_bytes()).unwrap();
+        c.1.shutdown(std::net::Shutdown::Write).unwrap();
+        let req = answer(&rx);
+        assert_eq!(req.params["params"]["text"], "once");
+        assert_eq!(c.recv()["ok"], true);
+        assert!(c.closed());
+        assert!(rx.try_recv().is_err());
+    }
+
     #[test]
     fn a_stale_key_file_is_replaced() {
         let dir = temp_dir("stale");
@@ -532,4 +656,52 @@ mod tests {
         }
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    fn boot_limits(limits: Limits) -> (u16, Receiver<ControlRequest>) {
+        let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
+        let port = listener.local_addr().unwrap().port();
+        (port, serve_with(listener, egui::Context::default(), KEY.to_string(), limits).unwrap())
+    }
+
+    fn small(wait_ms: u64, max_unauthenticated: usize) -> Limits {
+        Limits { first: FIRST_REQUEST, wait: Duration::from_millis(wait_ms), max_unauthenticated }
+    }
+
+    #[test]
+    fn a_request_that_timed_out_in_the_queue_never_runs() {
+        let (port, rx) = boot_limits(small(150, 8));
+        let mut c = Conn::open(port);
+        let r = c.call(&format!(r#"{{"id":1,"key":"{KEY}","method":"engine.execute","params":{{"command":"text.insert"}}}}"#));
+        assert_eq!(r["ok"], false);
+        assert!(r["error"].as_str().unwrap_or("").starts_with("timeout before execution"), "{r}");
+        // The UI gets to it late: it must not run it.
+        let req = rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        assert!(!req.start(), "a cancelled request was started");
+    }
+
+    #[test]
+    fn a_request_that_started_before_the_timeout_reports_an_unknown_outcome() {
+        let (port, rx) = boot_limits(small(150, 8));
+        let mut c = Conn::open(port);
+        c.send(&format!(r#"{{"id":1,"key":"{KEY}","method":"engine.execute","params":{{"command":"text.insert"}}}}"#));
+        let req = rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        assert!(req.start());
+        let r = c.recv();
+        assert!(r["error"].as_str().unwrap_or("").contains("outcome unknown"), "{r}");
+    }
+
+    #[test]
+    fn unauthenticated_connections_are_limited() {
+        let (port, rx) = boot_limits(small(5_000, 2));
+        let mut idle = [Conn::open(port), Conn::open(port)];
+        // Give the server time to count both.
+        std::thread::sleep(Duration::from_millis(200));
+        assert!(Conn::open(port).closed(), "the third idle connection was admitted");
+        // One authenticates: its slot is free again.
+        assert_eq!(idle[0].call(&format!(r#"{{"id":1,"method":"auth","key":"{KEY}"}}"#))["ok"], true);
+        std::thread::sleep(Duration::from_millis(200));
+        let mut again = Conn::open(port);
+        assert_eq!(again.call(&format!(r#"{{"id":1,"method":"auth","key":"{KEY}"}}"#))["ok"], true);
+        assert!(rx.try_recv().is_err());
+    }
 }

crates/mcp/src/backend.rs

@@ -184,48 +184,64 @@ impl Remote {
         Ok(req.to_string())
     }
 
-    fn roundtrip(&mut self, line: &str) -> std::io::Result<String> {
+    /// Send one line and read its reply. Once a write is attempted, any failure leaves the
+    /// outcome unknown: even a failed write may have delivered a complete request at EOF.
+    fn roundtrip(&mut self, line: &str) -> Result<String, Failure> {
         let Some((reader, writer)) = self.conn.as_mut() else {
-            return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected"));
+            return Err(Failure::NotSent(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected")));
         };
-        writer.write_all(line.as_bytes())?;
-        writer.write_all(b"\n")?;
-        writer.flush()?;
-        match read_bounded_line(reader, MAX_REPLY_LINE)? {
-            BoundedLine::Line(reply) => Ok(reply),
-            BoundedLine::Eof => Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed")),
-            BoundedLine::TooLong => Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "reply from app too long")),
-        }
+        roundtrip_io(reader, writer, line)
     }
 
-    /// One request and its reply, with one retry on a new connection (the app may have
-    /// restarted; a new connection also reads the key again).
+    /// One request and its reply. A request that was not sent, or one that only reads, is tried
+    /// once more on a new connection (the app may have restarted; a new connection also reads
+    /// the key again). A mutation whose outcome is unknown is never sent again: it may have run.
     fn exchange(&mut self, id: u64, method: &str, params: &Value) -> Result<Value, String> {
+        self.exchange_with(id, method, params, Self::roundtrip)
+    }
+
+    fn exchange_with(
+        &mut self,
+        id: u64,
+        method: &str,
+        params: &Value,
+        mut roundtrip: impl FnMut(&mut Self, &str) -> Result<String, Failure>,
+    ) -> Result<Value, String> {
         let unreachable = |addr: &str, e: std::io::Error, connected: bool| {
             let retry = if connected { String::new() } else { "; will retry on each call".to_string() };
             format!("WordCraft app at {addr} is not reachable: {e}{retry}")
         };
         let mut first = true;
         let reply = loop {
-            let tried = if self.conn.is_some() { Ok(()) } else { self.reconnect() };
+            let tried = if self.conn.is_some() { Ok(()) } else { self.reconnect().map_err(Failure::NotSent) };
             let result = match tried {
                 Ok(()) => {
                     let line = self.request_line(id, method, params)?;
-                    self.roundtrip(&line)
+                    roundtrip(self, &line)
                 }
                 Err(e) => Err(e),
             };
             match result {
                 Ok(r) => break r,
-                Err(e) => {
+                Err(f) => {
                     self.conn = None;
                     // A reply that arrived but can't be read (too long, not UTF-8): the app has
                     // run the request, so sending it again could run a state-changing one twice.
-                    if e.kind() == std::io::ErrorKind::InvalidData {
+                    if let Failure::Unknown(e) = &f
+                        && e.kind() == std::io::ErrorKind::InvalidData
+                    {
                         return Err(format!("bad reply from app: {e}"));
                     }
+                    if let Failure::Unknown(e) = &f
+                        && !is_read_only(method, params)
+                    {
+                        return Err(format!(
+                            "outcome unknown: the connection to the WordCraft app at {} failed while sending `{method}` or waiting for its reply ({e}); it may have been executed, so it is not sent again. Inspect the document to check",
+                            self.addr
+                        ));
+                    }
                     if !first {
-                        return Err(unreachable(&self.addr, e, self.connected));
+                        return Err(unreachable(&self.addr, f.into_error(), self.connected));
                     }
                     first = false;
                 }
@@ -235,6 +251,44 @@ impl Remote {
     }
 }
 
+/// Shared socket exchange, also used with a writer that fails after partial delivery in tests.
+fn roundtrip_io(reader: &mut impl std::io::BufRead, writer: &mut impl Write, line: &str) -> Result<String, Failure> {
+    let mut out = Vec::with_capacity(line.len() + 1);
+    out.extend_from_slice(line.as_bytes());
+    out.push(b'\n');
+    writer.write_all(&out).and_then(|()| writer.flush()).map_err(Failure::Unknown)?;
+    match read_bounded_line(reader, MAX_REPLY_LINE).map_err(Failure::Unknown)? {
+        BoundedLine::Line(reply) => Ok(reply),
+        BoundedLine::Eof => Err(Failure::Unknown(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed"))),
+        BoundedLine::TooLong => Err(Failure::Unknown(std::io::Error::new(std::io::ErrorKind::InvalidData, "reply from app too long"))),
+    }
+}
+
+/// Why a round trip failed.
+enum Failure {
+    /// No write was attempted: the app did not run it.
+    NotSent(std::io::Error),
+    /// After the request was sent: the app may have run it.
+    Unknown(std::io::Error),
+}
+
+impl Failure {
+    fn into_error(self) -> std::io::Error {
+        match self {
+            Self::NotSent(e) | Self::Unknown(e) => e,
+        }
+    }
+}
+
+/// Methods that only read: sending them twice does no harm.
+fn is_read_only(method: &str, params: &Value) -> bool {
+    match method {
+        "document.inspect" | "engine.commands" | "ui.inspect" | "ui.parity" => true,
+        "ui.render" | "ui.screenshot" => params.get("path").and_then(Value::as_str).is_none(),
+        _ => false,
+    }
+}
+
 impl Backend for Remote {
     fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
         let id = self.next_id;
@@ -498,4 +552,100 @@ mod tests {
         assert!(e.contains("not the app's address") && e.contains("not sent"), "{e}");
         assert!(seen.recv_timeout(WAIT).unwrap().get("key").is_none());
     }
+
+    #[test]
+    fn a_mutation_with_a_lost_reply_is_not_sent_again() {
+        // The app runs the request, then the connection drops without a reply.
+        let (port, seen) = fake_app(|_| None);
+        let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap();
+        let e = r.call("engine.execute", json!({"command": "text.insert", "params": {"text": "x"}})).unwrap_err();
+        assert!(e.contains("outcome unknown"), "{e}");
+        assert_eq!(seen.recv_timeout(WAIT).unwrap()["method"], "engine.execute");
+        std::thread::sleep(Duration::from_millis(200));
+        assert!(seen.try_recv().is_err(), "the mutation was sent a second time");
+    }
+
+    #[test]
+    fn a_read_with_a_lost_reply_is_sent_again() {
+        let n = Arc::new(AtomicUsize::new(0));
+        let (port, seen) =
+            fake_app(move |_| if n.fetch_add(1, Ordering::SeqCst) == 0 { None } else { Some((json!({"ok": true, "result": 3}), false)) });
+        let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap();
+        assert_eq!(r.call("document.inspect", json!({})).unwrap(), json!(3));
+        assert_eq!(seen.try_iter().count(), 2);
+    }
+
+    #[test]
+    fn a_failed_write_after_delivering_json_never_replays_a_mutation() {
+        // Deliver a complete request, then fail either its newline write or its flush. The
+        // stand-in, like the production server, accepts a complete JSON body at EOF.
+        struct FailingWriter<'a> {
+            socket: &'a mut TcpStream,
+            fail_flush: bool,
+            wrote: bool,
+        }
+        impl Write for FailingWriter<'_> {
+            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+                if self.wrote {
+                    return Err(std::io::ErrorKind::BrokenPipe.into());
+                }
+                self.wrote = true;
+                let len = buf.len() - usize::from(!self.fail_flush);
+                self.socket.write_all(&buf[..len])?;
+                self.socket.shutdown(std::net::Shutdown::Write)?;
+                Ok(len)
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Err(std::io::ErrorKind::BrokenPipe.into())
+            }
+        }
+        for fail_flush in [false, true] {
+            let (port, seen) = fake_app(|_| None);
+            let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap();
+            let mut attempts = 0;
+            let e = r
+                .exchange_with(1, "engine.execute", &json!({"command": "text.insert", "params": {"text": "once"}}), |r, line| {
+                    attempts += 1;
+                    let (reader, socket) = r.conn.as_mut().unwrap();
+                    roundtrip_io(reader, &mut FailingWriter { socket, fail_flush, wrote: false }, line)
+                })
+                .unwrap_err();
+            assert!(e.contains("outcome unknown"), "{e}");
+            assert_eq!(attempts, 1, "a failed write must not trigger another exchange");
+            assert_eq!(seen.recv_timeout(WAIT).unwrap()["params"]["params"]["text"], "once");
+            assert!(seen.try_recv().is_err());
+        }
+    }
+
+    #[test]
+    fn image_exports_with_a_lost_reply_are_not_replayed_but_in_memory_reads_are() {
+        for method in ["ui.render", "ui.screenshot"] {
+            for path in [json!("out.png"), json!(""), Value::Null] {
+                let count = Arc::new(AtomicUsize::new(0));
+                let n = count.clone();
+                let (port, seen) =
+                    fake_app(move |_| if n.fetch_add(1, Ordering::SeqCst) == 0 { None } else { Some((json!({"ok": true, "result": 3}), false)) });
+                let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap();
+                let result = r.call(method, json!({"path": path}));
+                if path.is_string() {
+                    assert!(result.unwrap_err().contains("outcome unknown"));
+                    assert_eq!(count.load(Ordering::SeqCst), 1);
+                } else {
+                    assert_eq!(result.unwrap(), json!(3));
+                    assert_eq!(count.load(Ordering::SeqCst), 2);
+                }
+                assert_eq!(seen.try_iter().count(), if path.is_string() { 1 } else { 2 });
+            }
+            assert!(is_read_only(method, &json!({})), "no path means PNG in the reply");
+        }
+    }
+
+    #[test]
+    fn a_mutation_not_sent_goes_out_on_a_new_connection() {
+        let (port, seen) = fake_app(|_| Some((json!({"ok": true, "result": 1}), false)));
+        let mut r = Remote::connect_with(&format!("127.0.0.1:{port}"), fixed_key("k")).unwrap();
+        r.conn = None;
+        assert_eq!(r.call("engine.execute", json!({"command": "file.new"})).unwrap(), json!(1));
+        assert_eq!(seen.try_iter().count(), 1);
+    }
 }

crates/mcp/src/server.rs

@@ -106,10 +106,14 @@ impl Server {
     /// the protocol stream).
     pub fn serve(&mut self, mut input: impl BufRead, mut output: impl Write) -> std::io::Result<()> {
         loop {
-            let reply = match read_bounded_line(&mut input, MAX_REQUEST_LINE)? {
-                BoundedLine::Eof => break,
-                BoundedLine::Line(line) => self.handle_line(&line),
-                BoundedLine::TooLong => Some(error(Value::Null, INVALID_REQUEST, "request line too long").to_string()),
+            let reply = match read_bounded_line(&mut input, MAX_REQUEST_LINE) {
+                Ok(BoundedLine::Eof) => break,
+                Ok(BoundedLine::Line(line)) => self.handle_line(&line),
+                Ok(BoundedLine::TooLong) => Some(error(Value::Null, INVALID_REQUEST, "request line too long").to_string()),
+                Err(e) if e.kind() == std::io::ErrorKind::InvalidData => {
+                    Some(error(Value::Null, PARSE_ERROR, "request line is not UTF-8").to_string())
+                }
+                Err(e) => return Err(e),
             };
             if let Some(reply) = reply {
                 output.write_all(reply.as_bytes())?;

crates/mcp/src/tests.rs

@@ -431,3 +431,24 @@ fn render_page_rejects_a_page_number_that_does_not_fit() {
     let r = call(&mut s, 31, "tools/call", json!({"name": "render_page", "arguments": {"page": 1, "scale": 0.5}}));
     assert_eq!(r["result"]["isError"], false, "page 1 still renders: {r}");
 }
+
+#[test]
+fn invalid_utf8_does_not_change_arguments_or_stop_later_requests() {
+    let mut s = Server::new(Box::new(Headless::default()));
+    tool(&mut s, "new_document", json!({}));
+    let mut input = br#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"type_text","arguments":{"text":"bad"}}}"#.to_vec();
+    let pos = input.windows(3).position(|w| w == b"bad").unwrap();
+    input[pos] = 0xff;
+    input.push(b'\n');
+    input.extend_from_slice(
+        b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\",\"params\":{\"name\":\"type_text\",\"arguments\":{\"text\":\"good\"}}}\n",
+    );
+    let mut out = Vec::new();
+    s.serve(std::io::Cursor::new(input), &mut out).unwrap();
+    let replies: Vec<Value> = String::from_utf8(out).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect();
+    assert_eq!(replies.len(), 2);
+    assert_eq!(replies[0]["error"]["code"], -32700);
+    assert_eq!(replies[1]["id"], 2);
+    assert_eq!(replies[1]["result"]["isError"], false);
+    assert_eq!(tool(&mut s, "get_text", json!({}))["text"], "good", "malformed text must never reach the backend");
+}

crates/ui-egui/src/app/control.rs

@@ -39,6 +39,11 @@ impl WordApp {
         let Some(rx) = self.control_rx.take() else { return };
         while let Ok(req) = rx.try_recv() {
             let reply = req.reply.clone();
+            if !req.start() {
+                // The caller timed out while this waited in the queue and was told it did not run.
+                log::debug!("control request {} dropped: its caller gave up", req.method);
+                continue;
+            }
             match control::handle(self, ctx, &req) {
                 control::Outcome::Done(v) => {
                     let _ = reply.send(v);

crates/ui-egui/src/control.rs

@@ -43,12 +43,45 @@ pub struct ControlRequest {
     pub method: String,
     pub params: Value,
     pub reply: Sender<Value>,
+    /// Shared with the waiter: whichever of "start" and "give up" happens first wins.
+    state: std::sync::Arc<std::sync::atomic::AtomicU8>,
+}
+
+const QUEUED: u8 = 0;
+const STARTED: u8 = 1;
+const CANCELLED: u8 = 2;
+
+/// The waiter's side of a [`ControlRequest`]: it gives up on a request that has not started.
+#[derive(Clone)]
+pub struct ControlWaiter(std::sync::Arc<std::sync::atomic::AtomicU8>);
+
+impl ControlWaiter {
+    /// Cancel the request. True when it had not started: it will never run. False when the UI
+    /// already started it: its outcome is unknown to the waiter.
+    pub fn cancel(&self) -> bool {
+        use std::sync::atomic::Ordering::SeqCst;
+        self.0.compare_exchange(QUEUED, CANCELLED, SeqCst, SeqCst).is_ok()
+    }
 }
 
 impl ControlRequest {
     pub fn new(method: impl Into<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<Value>) {
+        let (req, rx, _) = Self::with_waiter(method, params);
+        (req, rx)
+    }
+
+    /// Like [`ControlRequest::new`], plus the handle the waiter uses to cancel a queued request.
+    pub fn with_waiter(method: impl Into<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<Value>, ControlWaiter) {
         let (tx, rx) = std::sync::mpsc::channel();
-        (Self { method: method.into(), params, reply: tx }, rx)
+        let state = std::sync::Arc::new(std::sync::atomic::AtomicU8::new(QUEUED));
+        (Self { method: method.into(), params, reply: tx, state: state.clone() }, rx, ControlWaiter(state))
+    }
+
+    /// The UI is about to run the request. False when the waiter already gave up on it: it
+    /// must not run.
+    pub fn start(&self) -> bool {
+        use std::sync::atomic::Ordering::SeqCst;
+        self.state.compare_exchange(QUEUED, STARTED, SeqCst, SeqCst).is_ok()
     }
 }
 

docs/control-protocol.md

@@ -1,5 +1,7 @@
 # Control protocol
 
+> **Last updated:** 2026-10-31 · #2596: accept-thread startup failure disables the server and removes its key file; MCP retry policy distinguishes file exports from in-memory images.
+
 Start the app with a control port: `wordcraft --control 7981` (or `WORDCRAFT_CONTROL_PORT=7981`).
 A missing or invalid port after `--control` (not 1–65535) prints an error and exits with status 2.
 It listens on `127.0.0.1` only. Send one JSON object per line, with the window's key (see
@@ -50,6 +52,21 @@ any other host it sends no key, says so, and the app refuses its requests: to dr
 another machine, forward its port to `127.0.0.1` (for example `ssh -L 7981:127.0.0.1:7981 host`)
 and set `WORDCRAFT_CONTROL_KEY` to that app's key.
 
+## Delivery and startup failures
+
+If the accept thread cannot start, startup reports that the control server is off and removes
+its newly written key file; it does not report a listening server. Failure to start a worker
+for one connection closes only that connection and releases its admission slot.
+
+The server can accept a complete JSON body at EOF even without a final newline. A client that
+encounters a write error must therefore treat delivery as unknown once it has attempted to
+send. The MCP bridge does not replay mutations in this case, or image exports with a string
+`path`; see [delivery failures](mcp.md#delivery-failures).
+
+A request that times out while still queued is cancelled and cannot run later. If execution
+has already started, the timeout reports an unknown outcome rather than promising cancellation.
+Unauthenticated connections are limited to 64 while they wait for their first authorised request.
+
 ## View › Window
 
 The app's native windows (one per document, or several on one document) take part in View ›

docs/mcp.md

@@ -1,5 +1,7 @@
 # MCP server
 
+> **Last updated:** 2026-10-31 · Delivery review fixes for #2596: unknown outcomes are not replayed, including image exports to a path; invalid UTF-8 returns a parse error without stopping later requests.
+
 WordCraft speaks the [Model Context Protocol](https://modelcontextprotocol.io) over stdio, so
 Claude and other agents can write and edit documents.
 
@@ -18,6 +20,22 @@ isn't open yet: tool calls return "not reachable … will retry on each call" an
 app's port opens (#4078), so a client that starts before WordCraft doesn't lose the server for
 its whole session.
 
+## Delivery failures
+
+The connected bridge retries a request once if no write was attempted (for example, a failed
+connection). Once sending starts, a write or flush failure can mean the app already received
+the complete JSON body, even if its newline was not delivered. A lost reply has the same
+uncertainty. Mutations then return an `outcome unknown` error and are never automatically
+resent; inspect the document before deciding whether to repeat the operation yourself.
+
+Read-only inspection and in-memory `ui.render` / `ui.screenshot` calls may retry once. With a
+string `path`, rendering or taking a screenshot writes a file and is not retried after an
+unknown outcome. Unreadable replies are reported as errors without replaying the request.
+
+On stdio, an invalid UTF-8 request line returns JSON-RPC parse error `-32700`. It never reaches
+the backend, and the server continues with the next line instead of substituting characters
+in text or paths.
+
 ## Scripts from the command line
 
 `wordcraft-cli run --script job.json` (`--script -` reads standard input) runs the commands in a