Control: a transport that gives up knows whether the command was dispatched (#3662) (#4256)

* Control: a timeout says whether the command ran (#3662)

A request still queued when its deadline passes is dropped before
dispatch, but it answered with the same text as the transport giving up
on a command that may still apply. The two now read differently: "timeout:
not run ... safe to re-send" and "timeout: no reply in 60s ... may still
apply", both keeping the `timeout` prefix machines match on.

* Control: a transport that gives up knows whether the command was dispatched (#3662)

The transport and the UI thread share the reply deadline, and the
transport's wait runs out first, so the UI thread's "not run" reply
almost never reached a client. A request now carries a dispatch flag
settled once between them: the UI thread claims it before running, the
transport abandons it when it stops waiting. The transport then replies
"not run" for a request that will never run, and "may still apply"
only for one that was dispatched; a request given up on is never run.

apps/photocraft/src/control_server.rs

@@ -12,7 +12,7 @@ use photocraft_automation::security::{
     ConnectionLimiter, LineRead, MAX_CONNECTIONS, MAX_REQUEST_BYTES, authentication_reply, configure_stream, read_bounded_line, repeat_auth_reply,
 };
 use photocraft_ui_egui::ControlRequest;
-use photocraft_ui_egui::control::{REPLY_DEADLINE, timeout_error};
+use photocraft_ui_egui::control::REPLY_DEADLINE;
 use serde_json::{Value, json};
 
 pub fn start(port: u16, token: String, ctx: egui::Context) -> Receiver<ControlRequest> {
@@ -93,12 +93,13 @@ fn serve(stream: TcpStream, token: &str, tx: Sender<ControlRequest>, ctx: egui::
                 let deadline = Instant::now() + REPLY_DEADLINE;
                 let (mut req, rrx) = ControlRequest::new(method, params);
                 req.deadline = Some(deadline);
+                let dispatch = req.dispatch.clone();
                 if tx.send(req).is_err() {
                     break;
                 }
                 ctx.request_repaint();
                 let mut r =
-                    rrx.recv_timeout(deadline.saturating_duration_since(Instant::now())).unwrap_or_else(|_| json!({"ok": false, "error": timeout_error()}));
+                    rrx.recv_timeout(deadline.saturating_duration_since(Instant::now())).unwrap_or_else(|_| json!({"ok": false, "error": dispatch.give_up()}));
                 if let Some(o) = r.as_object_mut() {
                     o.insert("id".into(), id);
                 }

crates/ui-egui/src/control.rs

@@ -46,12 +46,22 @@ pub type ControlResponse = Value;
 /// The control server's per-request reply deadline.
 pub const REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(60);
 
-/// The error both timeout sites return when a request's [`REPLY_DEADLINE`] passes: the
-/// transport thread waiting for a reply, and the UI thread rejecting a request it is only
-/// now draining. Built from the same constant, so the text cannot drift from the deadline.
+/// The error the transport thread returns when a dispatched command gave no reply within
+/// [`REPLY_DEADLINE`]. It warns that the command may still apply: re-sending blindly could apply
+/// an edit twice (#3662). Built from the deadline constant, so the text cannot drift from it.
 pub fn timeout_error() -> String {
     format!(
-        "timeout: no reply in {}s — the command may still be running, or the app was not drawing frames (display asleep, window fully occluded, or a long operation blocked the frame loop)",
+        "timeout: no reply in {}s — the command may still be running and may still apply, or the app was not drawing frames (display asleep, window fully occluded, or a long operation blocked the frame loop)",
+        REPLY_DEADLINE.as_secs()
+    )
+}
+
+/// The error for a request still queued when its deadline passed, from the UI thread or the
+/// transport (see [`Dispatch`]). It was dropped before dispatch, so re-sending it is safe (#3662). Same `timeout` prefix as
+/// [`timeout_error`], which machines match on.
+pub fn expired_error() -> String {
+    format!(
+        "timeout: not run — the request was still queued when its {}s deadline passed, so it was dropped before dispatch and is safe to re-send; the app was not drawing frames (display asleep, window fully occluded, or a long operation blocked the frame loop)",
         REPLY_DEADLINE.as_secs()
     )
 }
@@ -62,12 +72,44 @@ pub struct ControlRequest {
     pub reply: Sender<ControlResponse>,
     /// Requests still queued at this instant are rejected; already dispatched work continues.
     pub deadline: Option<std::time::Instant>,
+    /// Shared with the transport, so that when it stops waiting it knows whether the command ran.
+    pub dispatch: Dispatch,
 }
 
 impl ControlRequest {
     pub fn new(method: impl Into<String>, params: Value) -> (Self, std::sync::mpsc::Receiver<ControlResponse>) {
         let (tx, rx) = std::sync::mpsc::channel();
-        (Self { method: method.into(), params, reply: tx, deadline: None }, rx)
+        (Self { method: method.into(), params, reply: tx, deadline: None, dispatch: Dispatch::default() }, rx)
+    }
+}
+
+/// Whether a request was dispatched, settled once between the UI thread (about to run it) and
+/// the transport (giving up on its reply). The transport's clock runs out no later than the UI
+/// thread's, so without this it could only ever say the command may still apply (#3662).
+#[derive(Clone, Default)]
+pub struct Dispatch(std::sync::Arc<std::sync::atomic::AtomicU8>);
+
+const QUEUED: u8 = 0;
+const DISPATCHED: u8 = 1;
+const ABANDONED: u8 = 2;
+
+impl Dispatch {
+    /// The UI thread claims the request to run it: false once the transport gave up on it. A
+    /// request the UI thread rejects as expired stays queued, which `give_up` reads as not run.
+    pub(crate) fn start(&self) -> bool {
+        self.settle(DISPATCHED) == DISPATCHED
+    }
+
+    /// The transport stops waiting: the error to reply with, [`expired_error`] when the command
+    /// can no longer run and [`timeout_error`] when it was dispatched and may still apply.
+    pub fn give_up(&self) -> String {
+        if self.settle(ABANDONED) == DISPATCHED { timeout_error() } else { expired_error() }
+    }
+
+    /// Moves a queued request to `to`; returns the state it ends in.
+    fn settle(&self, to: u8) -> u8 {
+        use std::sync::atomic::Ordering;
+        self.0.compare_exchange(QUEUED, to, Ordering::AcqRel, Ordering::Acquire).map_or_else(|state| state, |_| to)
     }
 }
 
@@ -1565,11 +1607,54 @@ mod tests {
         retry.deadline = Some(Instant::now() + Duration::from_secs(60));
         tx.send(retry).unwrap();
         app.drain_control(&egui::Context::default());
-        assert_eq!(expired_reply.try_recv().unwrap(), json!({"ok": false, "error": super::timeout_error()}));
+        let expired_error = expired_reply.try_recv().unwrap()["error"].as_str().unwrap().to_string();
         assert_eq!(retry_reply.try_recv().unwrap()["ok"], true);
         let after = app.session.active().unwrap();
         assert_eq!(after.doc.layers.len(), layers + 1, "only the live retry adds a layer");
         assert_eq!(after.history.past_len(), steps + 1, "the expired edit adds no undo step");
+        // #3662: a caller must be able to tell this reply, safe to re-send, from the
+        // transport's, sent while the command may still apply.
+        assert!(expired_error.starts_with("timeout"), "machines match on the prefix: {expired_error}");
+        assert!(expired_error.contains("not run"), "the expired request says it never ran: {expired_error}");
+        let transport_error = super::timeout_error();
+        assert!(!transport_error.contains("not run"), "the transport can't know it didn't run: {transport_error}");
+        assert!(transport_error.contains("may still apply"), "the transport warns against a blind re-send: {transport_error}");
+    }
+
+    #[test]
+    fn a_transport_that_gives_up_learns_whether_the_command_can_still_run() {
+        // #3662: the transport's clock runs out no later than the UI thread's, so it is the
+        // one that answers; it must still tell a request that will never run from one that may.
+        use std::time::{Duration, Instant};
+        let mut app = PhotocraftApp::new(photocraft_engine::Session::new(), crate::Services::default());
+        app.run("file.new", json!({"width": 8, "height": 8})).unwrap();
+        let layers = app.session.active().unwrap().doc.layers.len();
+        let (tx, rx) = std::sync::mpsc::channel();
+        app = app.with_control(rx);
+        let ctx = egui::Context::default();
+        let request = |deadline: Instant| {
+            let (mut req, reply) = ControlRequest::new("engine.execute", json!({"command": "layer.new.layer"}));
+            req.deadline = Some(deadline);
+            (req.dispatch.clone(), req, reply)
+        };
+
+        let (dispatch, queued, _) = request(Instant::now() + Duration::from_secs(60));
+        tx.send(queued).unwrap();
+        assert_eq!(dispatch.give_up(), super::expired_error(), "still queued: it will not run");
+        app.drain_control(&ctx);
+        assert_eq!(app.session.active().unwrap().doc.layers.len(), layers, "a request given up on never runs");
+
+        let (dispatch, expired, _) = request(Instant::now() - Duration::from_secs(1));
+        tx.send(expired).unwrap();
+        app.drain_control(&ctx);
+        assert_eq!(dispatch.give_up(), super::expired_error(), "rejected by the UI thread: it did not run");
+
+        let (dispatch, live, reply) = request(Instant::now() + Duration::from_secs(60));
+        tx.send(live).unwrap();
+        app.drain_control(&ctx);
+        assert_eq!(reply.try_recv().unwrap()["ok"], true);
+        assert_eq!(app.session.active().unwrap().doc.layers.len(), layers + 1);
+        assert_eq!(dispatch.give_up(), super::timeout_error(), "dispatched: it may still apply");
     }
 
     #[test]

crates/ui-egui/src/lib.rs

@@ -1510,8 +1510,8 @@ impl PhotocraftApp {
         let Some(rx) = self.control_rx.take() else { return };
         while let Ok(req) = rx.try_recv() {
             let reply = req.reply.clone();
-            if req.deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
-                let _ = reply.send(serde_json::json!({"ok": false, "error": control::timeout_error()}));
+            if req.deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) || !req.dispatch.start() {
+                let _ = reply.send(serde_json::json!({"ok": false, "error": control::expired_error()}));
                 continue;
             }
             match control::handle(self, ctx, &req) {

docs/control-protocol.md

@@ -38,7 +38,7 @@ Each reply is one JSON line with the same `id`:
 {"id": 2, "ok": false, "error": "unknown tool `foo`"}
 ```
 
-The desktop server waits up to 60 seconds for a reply. A request still queued at that deadline is rejected before dispatch; a timeout does not cancel work that has already started. The timeout error names the likely causes: the command may still be running, or the request was queued while the window was not drawing frames — on Wayland the frame update (and with it the control drain) only runs on compositor frame callbacks, so a display that is asleep or a fully occluded window makes every request time out while the app itself is healthy.
+The desktop server waits up to 60 seconds for a reply. A request still queued at that deadline is rejected before dispatch; a timeout does not cancel work that has already started. The two timeouts read differently, both starting with `timeout`: a request that was never dispatched (still queued at the deadline) says `timeout: not run` and is safe to re-send, while `timeout: no reply in 60s` means the command was dispatched and may still be running and still apply, so check the document before re-sending it. The timeout error names the likely causes: the command may still be running, or the request was queued while the window was not drawing frames — on Wayland the frame update (and with it the control drain) only runs on compositor frame callbacks, so a display that is asleep or a fully occluded window makes every request time out while the app itself is healthy.
 
 The transport is `apps/photocraft/src/control_server.rs`, and the handlers are in `crates/ui-egui/src/control.rs`. The MCP server (`photocraft-cli mcp --bridge 127.0.0.1:<port>`, crate `photocraft-automation`) wraps this same protocol. See [MCP bridge](#mcp-bridge) below.