Merge remote-tracking branch 'origin/main' into fix/issue-1233-atomic-write-temp

# Conflicts:
#	crates/engine/src/config.rs
#	crates/engine/src/tests_settings.rs

crates/automation/src/bridge.rs

@@ -1,5 +1,5 @@
 //! Blocking client for the desktop app's JSON-lines control channel (`docs/control-protocol.md`).
-//! One persistent loopback connection, reconnected once if the app went away.
+//! One persistent loopback connection. Only explicitly read-only queries can be resent.
 
 use std::io::{BufRead, BufReader, Write};
 use std::net::{TcpStream, ToSocketAddrs};
@@ -50,16 +50,20 @@ impl BridgeClient {
 
     /// Call a control method; returns the reply's `result`, or the app's `error` as [`Error::App`].
     pub fn call(&mut self, method: &str, params: Value) -> Result<Value> {
+        let retry_read =
+            matches!(method, "engine.commands" | "ui.menu.list" | "ui.inspect" | "ui.elements" | "ui.viewer.locate" | "ui.viewer.hit" | "ui.timeline.locate")
+                || (method == "ui.playback" && params["action"] == "status");
+        let unknown = |error| Error::Bridge(format!("`{method}` request outcome is unknown: {error}; inspect the app before retrying"));
         let mut last = Error::Bridge("not attempted".into());
         for attempt in 0..2 {
             if self.conn.is_none() {
                 self.conn = Some(self.connect()?);
             }
             let id = self.next_id;
-            self.next_id += 1;
+            self.next_id = self.next_id.checked_add(1).ok_or_else(|| Error::Bridge("control request id space exhausted".into()))?;
             let Some((r, w)) = self.conn.as_mut() else { continue };
             let line = format!("{}\n", json!({"id": id, "method": method, "params": params}));
-            // `Ok(Err(_))`: the connection is dead before a reply (safe to resend once);
+            // `Ok(Err(_))`: transmission/reply failed (the app may have applied the request);
             // `Err(_)`: a timeout or bad reply (never resent: the app may have run the request).
             let res = (|| -> Result<std::result::Result<Value, Error>> {
                 if let Err(e) = w.write_all(line.as_bytes()) {
@@ -80,8 +84,11 @@ impl BridgeClient {
                 Ok(Ok(v)) if v.get("ok").and_then(Value::as_bool) == Some(true) => return Ok(v.get("result").cloned().unwrap_or(Value::Null)),
                 Ok(Ok(v)) => return Err(Error::App(v.get("error").and_then(Value::as_str).unwrap_or("error").to_string())),
                 Ok(Err(e)) => {
-                    // Stale connection (app restarted): reconnect and resend once.
                     self.conn = None;
+                    if !retry_read {
+                        return Err(unknown(e));
+                    }
+                    // A read-only query can be repeated after a stale connection or lost reply.
                     last = e;
                     if attempt == 1 {
                         break;
@@ -90,7 +97,7 @@ impl BridgeClient {
                 Err(e) => {
                     // The stream may still deliver the late reply: start fresh next call.
                     self.conn = None;
-                    return Err(e);
+                    return Err(if retry_read { e } else { unknown(e) });
                 }
             }
         }

crates/automation/src/tests.rs

@@ -511,6 +511,128 @@ fn bridge_rejects_non_loopback() {
     assert!(Backend::bridge("localhost:9877").is_ok());
 }
 
+/// The app applies the first request but loses its reply. The next connection can be either
+/// an automatic retry or an explicit follow-up; the returned counts distinguish them.
+fn lost_reply_server(first_reply: Option<&'static str>) -> (String, std::thread::JoinHandle<()>) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let worker = std::thread::spawn(move || {
+        let mut session = Session::default();
+        let mut mutations = 0;
+        for attempt in 0..2 {
+            let (stream, _) = listener.accept().unwrap();
+            stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+            let mut output = stream.try_clone().unwrap();
+            let mut input = String::new();
+            BufReader::new(stream).read_line(&mut input).unwrap();
+            let request: Value = serde_json::from_str(&input).unwrap();
+            let result = if request["method"] == "engine.execute" {
+                mutations += 1;
+                session.execute(request["params"]["command"].as_str().unwrap(), request["params"]["params"].clone()).unwrap()
+            } else if request["method"] == "ui.inspect" || (request["method"] == "ui.playback" && request["params"]["action"] == "status") {
+                json!({"mutations": mutations, "requests": attempt + 1, "comps": session.project.comps().count()})
+            } else {
+                mutations += 1;
+                json!({"mutations": mutations})
+            };
+            if attempt == 1 {
+                writeln!(output, "{}", json!({"id": request["id"], "ok": true, "result": result})).unwrap();
+            } else if let Some(reply) = first_reply {
+                writeln!(output, "{reply}").unwrap();
+            }
+        }
+    });
+    (addr, worker)
+}
+
+#[test]
+fn bridge_does_not_replay_a_mutation_when_its_reply_is_lost() {
+    let (addr, worker) = lost_reply_server(None);
+    let mut backend = Backend::bridge(&addr).unwrap();
+    let error = backend.exec("comp.new", json!({"width": 64, "height": 48, "duration": 1})).unwrap_err();
+    assert!(error.to_string().contains("outcome is unknown"), "{error}");
+    let state = backend.control("ui.inspect", json!({})).unwrap();
+    assert_eq!(state["mutations"], 1, "the first command was applied exactly once");
+    assert_eq!(state["comps"], 1);
+    worker.join().unwrap();
+}
+
+#[test]
+fn bridge_retries_read_only_queries_after_a_lost_reply() {
+    let (addr, worker) = lost_reply_server(None);
+    let mut backend = Backend::bridge(&addr).unwrap();
+    let state = backend.control("ui.inspect", json!({})).unwrap();
+    assert_eq!(state["requests"], 2);
+    assert_eq!(state["mutations"], 0);
+    worker.join().unwrap();
+}
+
+#[test]
+fn bridge_reports_unknown_outcome_after_a_malformed_mutation_reply() {
+    let (addr, worker) = lost_reply_server(Some("{invalid reply"));
+    let mut backend = Backend::bridge(&addr).unwrap();
+    let error = backend.exec("comp.new", json!({"width": 64, "height": 48, "duration": 1})).unwrap_err();
+    let message = error.to_string();
+    assert!(message.contains("outcome is unknown") && message.contains("bad reply"), "{message}");
+    assert_eq!(backend.control("ui.inspect", json!({})).unwrap()["comps"], 1);
+    worker.join().unwrap();
+}
+
+#[test]
+fn bridge_does_not_assume_an_unknown_method_is_read_only() {
+    let (addr, worker) = lost_reply_server(None);
+    let mut backend = Backend::bridge(&addr).unwrap();
+    let error = backend.control("future.customOperation", json!({})).unwrap_err();
+    assert!(error.to_string().contains("outcome is unknown"), "{error}");
+    assert_eq!(backend.control("ui.inspect", json!({})).unwrap()["mutations"], 1);
+    worker.join().unwrap();
+}
+
+#[test]
+fn bridge_playback_edits_and_default_toggle_are_not_replayed() {
+    for params in [json!({}), json!({"action": "toggle"}), json!({"action": "play"}), json!({"action": "stop"})] {
+        let (addr, worker) = lost_reply_server(None);
+        let mut backend = Backend::bridge(&addr).unwrap();
+        let error = backend.control("ui.playback", params).unwrap_err();
+        assert!(error.to_string().contains("outcome is unknown"), "{error}");
+        assert_eq!(backend.control("ui.inspect", json!({})).unwrap()["mutations"], 1);
+        worker.join().unwrap();
+    }
+}
+
+#[test]
+fn bridge_playback_status_can_be_retried() {
+    let (addr, worker) = lost_reply_server(None);
+    let mut backend = Backend::bridge(&addr).unwrap();
+    let state = backend.control("ui.playback", json!({"action": "status"})).unwrap();
+    assert_eq!(state["requests"], 2);
+    assert_eq!(state["mutations"], 0);
+    worker.join().unwrap();
+}
+
+#[test]
+fn bridge_connect_failure_can_be_retried_when_the_app_starts() {
+    let reserved = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = reserved.local_addr().unwrap();
+    drop(reserved);
+    let mut backend = Backend::bridge(&addr.to_string()).unwrap();
+    let error = backend.control("ui.inspect", json!({})).unwrap_err();
+    assert!(error.to_string().contains("cannot connect"), "{error}");
+    assert!(!error.to_string().contains("outcome is unknown"), "nothing was sent");
+    let listener = TcpListener::bind(addr).unwrap();
+    let worker = std::thread::spawn(move || {
+        let (stream, _) = listener.accept().unwrap();
+        stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+        let mut output = stream.try_clone().unwrap();
+        let mut input = String::new();
+        BufReader::new(stream).read_line(&mut input).unwrap();
+        let request: Value = serde_json::from_str(&input).unwrap();
+        writeln!(output, "{}", json!({"id": request["id"], "ok": true, "result": {"ready": true}})).unwrap();
+    });
+    assert_eq!(backend.control("ui.inspect", json!({})).unwrap()["ready"], true);
+    worker.join().unwrap();
+}
+
 #[test]
 fn script_ui_and_history_tools() {
     let mut sess = Session::default();

crates/engine/src/config.rs

@@ -164,48 +164,50 @@ impl ConfigStore for DirConfig {
     }
 }
 
-/// The temporary file an atomic write of `path` goes through. Unique per call (process id plus a
-/// counter), so two writers saving the same destination never share a temporary — one cannot
-/// truncate, rename or clean up another's — and there is no predictable name for a pre-existing
-/// file or symlink to sit at.
-pub fn temp_path(path: &Path) -> PathBuf {
+/// A new, empty temporary file next to `path` for an atomic write of it, and its path. The name
+/// is unique to this process and this write (`.effectcraft-<pid>-<n>.tmp`) and the file is created
+/// exclusively, so two writers never share one and nothing already at that name (a file, a
+/// symbolic link) is opened or followed (#1233). It ends in `.tmp`, so a file left by a hard
+/// kill never counts as an auto-save. Its bounded name does not include the destination's name,
+/// so a valid long destination name still fits the filesystem's component limit.
+fn create_temp(path: &Path) -> std::io::Result<(std::fs::File, PathBuf)> {
     use std::sync::atomic::{AtomicU64, Ordering};
     static NEXT: AtomicU64 = AtomicU64::new(0);
-    let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
-    name.push(format!(".{}-{}.tmp", std::process::id(), NEXT.fetch_add(1, Ordering::Relaxed)));
-    path.with_file_name(name)
+    let mut last = None;
+    for _ in 0..16 {
+        let n = NEXT.fetch_add(1, Ordering::Relaxed);
+        let tmp = path.with_file_name(format!(".effectcraft-{}-{n}.tmp", std::process::id()));
+        match std::fs::OpenOptions::new().write(true).create_new(true).open(&tmp) {
+            Ok(f) => return Ok((f, tmp)),
+            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => last = Some(e),
+            Err(e) => return Err(e),
+        }
+    }
+    Err(last.unwrap_or_else(|| std::io::Error::other("no free temporary file name")))
 }
 
 /// Write `data` to `path` atomically: write a temporary file next to it, flush it to disk, then
 /// rename it over `path`. A crash at any point leaves either the old file or the new one, never a
-/// torn file.
+/// torn file, and concurrent writers each publish a whole file of their own.
 pub fn atomic_write(path: &Path, data: &[u8]) -> std::io::Result<()> {
-    atomic_write_with(path, data, |tmp, data| {
+    atomic_write_with(path, data, |f, data| {
         use std::io::Write;
-        // Exclusive creation: an entry already at the temporary's name (a leftover from a crashed
-        // writer, or something planted there) is never opened, truncated or written through.
-        let mut f = std::fs::OpenOptions::new().write(true).create_new(true).open(tmp)?;
         f.write_all(data)?;
         f.sync_all()
     })
 }
 
-/// [`atomic_write`] with an injectable temp-file writer (tests simulate a crash mid-write). The
-/// temporary belongs to this call alone: the writer must create it exclusively, and a failed
-/// write removes only its own temporary and leaves `path` untouched. An `AlreadyExists` from the
-/// writer means the name was taken (never opened by us): that entry is left alone and another
-/// unique name is tried.
-pub fn atomic_write_with(path: &Path, data: &[u8], mut write: impl FnMut(&Path, &[u8]) -> std::io::Result<()>) -> std::io::Result<()> {
-    for _ in 0..16 {
-        let tmp = temp_path(path);
-        match write(&tmp, data) {
-            Ok(()) => return std::fs::rename(&tmp, path),
-            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
-            Err(e) => {
-                let _ = std::fs::remove_file(&tmp);
-                return Err(e);
-            }
-        }
+/// [`atomic_write`] with an injectable temp-file writer (tests simulate a crash mid-write); it is
+/// given the temporary file, already created. A failed write or rename removes the temporary
+/// file and leaves `path` untouched.
+pub fn atomic_write_with(path: &Path, data: &[u8], write: impl FnOnce(&mut std::fs::File, &[u8]) -> std::io::Result<()>) -> std::io::Result<()> {
+    let (mut f, tmp) = create_temp(path)?;
+    let written = write(&mut f, data);
+    // Closed before the rename (Windows can't rename an open file).
+    drop(f);
+    if let Err(e) = written.and_then(|()| std::fs::rename(&tmp, path)) {
+        let _ = std::fs::remove_file(&tmp);
+        return Err(e);
     }
-    Err(std::io::Error::new(std::io::ErrorKind::AlreadyExists, format!("{}: no free temporary name", path.display())))
+    Ok(())
 }

crates/engine/src/lib.rs

@@ -495,6 +495,8 @@ pub struct Session {
     /// Offloaded jobs that ended, newest last (kind, error): how a caller waiting for one
     /// (`wait: true` over the control channel in the browser) learns its outcome.
     pub offload_log: Vec<(offload::JobKind, Option<String>)>,
+    /// Number of older completion entries evicted from the bounded offload log.
+    pub offload_log_start: usize,
     /// Generic background tasks (Content-Aware Fill, Scene Edit Detection), see [`jobs`].
     pub tasks: Vec<jobs::Task>,
     /// Finished tasks, newest last (Progress panel, `jobs.list`).
@@ -598,6 +600,7 @@ impl Default for Session {
             offload: None,
             offloaded: vec![],
             offload_log: vec![],
+            offload_log_start: 0,
             browser: None,
             storage: None,
             tasks: vec![],

crates/engine/src/offload.rs

@@ -171,6 +171,20 @@ pub enum JobKind {
 }
 
 impl JobKind {
+    /// The corresponding `jobs.list` id, also used by waiting control commands.
+    pub fn job_id(self) -> &'static str {
+        match self {
+            JobKind::Render => "render",
+            JobKind::Warp => "warp",
+            JobKind::WarpPlan => "warpPlan",
+            JobKind::Camera => "camera",
+            JobKind::RotoFreeze | JobKind::RotoPropagate => "roto",
+            JobKind::Track => "track",
+            JobKind::MaskTrack => "maskTrack",
+            JobKind::ContentFill => "contentFill",
+        }
+    }
+
     /// The undo step an analysis result becomes.
     pub fn label(self) -> &'static str {
         match self {
@@ -699,6 +713,9 @@ pub struct OffloadedJob {
     /// Segmentations received (Roto Brush propagation).
     pub segs: u64,
     started: web_time::Instant,
+    /// The analysed group as the job started: its result is merged with what changed since
+    /// ([`merge_group`]).
+    base: Option<Box<PropGroup>>,
 }
 
 /// A Render Queue job running in a worker (see [`crate::render_queue::RenderJob`]).
@@ -730,6 +747,7 @@ impl Session {
         }
         let id = next_id();
         let inbox = Arc::new(Inbox::default());
+        let base = self.project.comp(comp).and_then(|c| c.layer(layer)).and_then(|l| l.props.find_group(uid)).cloned().map(Box::new);
         let req = WorkerRequest { id, project: self.project.to_json(), files: footage_files(&self.project), job };
         off.start(req, inbox.clone())?;
         self.offloaded.push(OffloadedJob {
@@ -742,6 +760,7 @@ impl Session {
             progress: OffloadProgress::default(),
             segs: 0,
             started: web_time::Instant::now(),
+            base,
         });
         self.events.push(Event::ProjectChanged { revision: self.revision });
         Ok(())
@@ -779,7 +798,7 @@ impl Session {
                         j.progress.done = done;
                         j.progress.total = total;
                     }
-                    WorkerReply::Group { comp, layer, group, message } => results.push((j.kind, comp, layer, group, message)),
+                    WorkerReply::Group { comp, layer, group, message } => results.push((j.id, j.kind, comp, layer, group, message, j.base.clone())),
                     WorkerReply::Fill { plan, files } => fills.push((plan, files)),
                     WorkerReply::WarpPlan { key, summary } => effectcraft_effects::warp_stab::store_summary(key, *summary),
                     WorkerReply::Segs { segs } => {
@@ -802,7 +821,7 @@ impl Session {
                 Err(e) => self.events.push(Event::Toast { message: format!("Content-Aware Fill: {e}"), error: true }),
             }
         }
-        for (kind, comp, layer, group, message) in results {
+        for (id, kind, comp, layer, group, message, base) in results {
             let uid = group.uid;
             let r = self.edit(kind.label(), None, |p, _| {
                 let g = p
@@ -810,12 +829,31 @@ impl Session {
                     .and_then(|c| c.layer_mut(layer))
                     .and_then(|l| l.props.find_group_mut(uid))
                     .ok_or_else(|| crate::EngineError::Other("the analysed property was removed during the analysis".into()))?;
-                *g = *group;
+                // Edits made while the job ran stay (#1249): only what the analysis changed is
+                // taken from its result.
+                *g = match base {
+                    Some(base) => merge_group(&base, g, &group).ok_or_else(|| {
+                        crate::EngineError::Other(format!(
+                            "“{}” was changed during the analysis where its result changes it too, so nothing was applied: analyse again",
+                            g.name
+                        ))
+                    })?,
+                    None => *group,
+                };
                 Ok(())
             });
             match r {
                 Ok(()) => self.events.push(Event::Toast { message, error: false }),
-                Err(e) => self.events.push(Event::Toast { message: format!("{}: {e}", kind.label()), error: true }),
+                Err(e) => {
+                    // Completion consumers read the job's error, not its toasts. Keep an apply
+                    // rejection on the originating job; Done logs and reports it just like a
+                    // WorkerReply::Failed, without failing another concurrent analysis.
+                    if let Some(job) = self.offloaded.iter_mut().find(|j| j.id == id) {
+                        job.progress.error = Some(e.to_string());
+                    } else {
+                        self.events.push(Event::Toast { message: format!("{}: {e}", kind.label()), error: true });
+                    }
+                }
             }
         }
         let mut done = vec![];
@@ -830,6 +868,7 @@ impl Session {
             self.offload_log.push((kind, p.error.clone().or_else(|| p.cancelled.then(|| "stopped".to_string()))));
             if self.offload_log.len() > 32 {
                 self.offload_log.remove(0);
+                self.offload_log_start = self.offload_log_start.saturating_add(1);
             }
             if let Some(e) = p.error {
                 self.events.push(Event::Toast { message: format!("{}: {e}", kind.label()), error: true });
@@ -880,6 +919,76 @@ impl Session {
     }
 }
 
+/// `live` if only it changed since `base`, `result` if only that did (or both the same way);
+/// `None` when both changed it differently.
+fn pick<T: PartialEq + Clone>(base: &T, live: &T, result: &T) -> Option<T> {
+    if live == base {
+        Some(result.clone())
+    } else if result == base || result == live {
+        Some(live.clone())
+    } else {
+        None
+    }
+}
+
+fn node_uid(n: &effectcraft_project::Node) -> Uid {
+    match n {
+        effectcraft_project::Node::Prop(p) => p.uid,
+        effectcraft_project::Node::Group(g) => g.uid,
+    }
+}
+
+/// Merge an analysis `result` into the `live` group (#1249). `base` is the group as the job
+/// started. The group's own fields and each property are taken from whichever side changed
+/// them, groups are merged child by child (matched by uid), and a child added or removed by one
+/// side is added or removed. `None` when both sides changed the same field or property
+/// differently, one side removed a child the other changed, or both reordered or added and
+/// removed children differently: a conflict.
+pub(crate) fn merge_group(base: &PropGroup, live: &PropGroup, result: &PropGroup) -> Option<PropGroup> {
+    use effectcraft_project::Node;
+    if live == base {
+        return Some(result.clone());
+    }
+    if result == base || result == live {
+        return Some(live.clone());
+    }
+    let header = |g: &PropGroup| PropGroup { children: vec![], ..g.clone() };
+    let mut out = pick(&header(base), &header(live), &header(result))?;
+    let ids = |g: &PropGroup| g.children.iter().map(node_uid).collect::<Vec<_>>();
+    let (bi, li, ri) = (ids(base), ids(live), ids(result));
+    // The children's order and set: from the side that changed them.
+    let order = if li == bi {
+        ri
+    } else if ri == bi || ri == li {
+        li
+    } else {
+        return None;
+    };
+    fn find(g: &PropGroup, u: Uid) -> Option<&effectcraft_project::Node> {
+        g.children.iter().find(|n| node_uid(n) == u)
+    }
+    // A child one side removed must be unchanged on the other, or that side's edit would be lost.
+    for u in bi.iter().filter(|u| !order.contains(u)) {
+        let b = find(base, *u);
+        if [find(live, *u), find(result, *u)].into_iter().flatten().any(|n| Some(n) != b) {
+            return None;
+        }
+    }
+    for u in order {
+        let node = match (find(base, u), find(live, u), find(result, u)) {
+            (Some(Node::Group(b)), Some(Node::Group(l)), Some(Node::Group(r))) => Node::Group(merge_group(b, l, r)?),
+            (Some(b), Some(l), Some(r)) => pick(b, l, r)?,
+            // Added by one side, or by both alike.
+            (None, Some(l), None) => l.clone(),
+            (None, None, Some(r)) => r.clone(),
+            (None, Some(l), Some(r)) if l == r => l.clone(),
+            _ => return None,
+        };
+        out.children.push(node);
+    }
+    Some(out)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -892,6 +1001,69 @@ mod tests {
         a
     }
 
+    /// A group with properties `a` (uid 2) and `b` (uid 3) and a sub-group (uid 4) holding `c`
+    /// (uid 5).
+    fn group() -> PropGroup {
+        use effectcraft_keyframe::Value as KV;
+        use effectcraft_project::{Node, Property};
+        let mut sub = PropGroup::new(4, "sub", "Sub");
+        sub.children.push(Node::Prop(Property::new(5, "c", "C", KV::Scalar(0.0))));
+        let mut g = PropGroup::new(1, "fx", "Effect");
+        g.children.push(Node::Prop(Property::new(2, "a", "A", KV::Scalar(0.0))));
+        g.children.push(Node::Prop(Property::new(3, "b", "B", KV::Scalar(0.0))));
+        g.children.push(Node::Group(sub));
+        g
+    }
+
+    fn set(g: &mut PropGroup, uid: Uid, v: f64) {
+        g.find_mut(uid).unwrap().value = effectcraft_keyframe::Value::Scalar(v);
+    }
+
+    fn get(g: &PropGroup, uid: Uid) -> f64 {
+        match g.find(uid).unwrap().value {
+            effectcraft_keyframe::Value::Scalar(v) => v,
+            ref v => panic!("{v:?}"),
+        }
+    }
+
+    #[test]
+    fn analysis_results_merge_with_edits_made_meanwhile() {
+        use effectcraft_project::Node;
+        let base = group();
+        // The user changes `a` and switches the effect off; the analysis writes `b` and `c`.
+        let mut live = base.clone();
+        set(&mut live, 2, 1.0);
+        live.enabled = false;
+        let mut result = base.clone();
+        set(&mut result, 3, 2.0);
+        set(&mut result, 5, 3.0);
+        let m = merge_group(&base, &live, &result).unwrap();
+        assert_eq!((get(&m, 2), get(&m, 3), get(&m, 5), m.enabled), (1.0, 2.0, 3.0, false));
+        // Unchanged on either side: the other side wins whole.
+        assert_eq!(merge_group(&base, &base, &result).unwrap(), result);
+        assert_eq!(merge_group(&base, &live, &base).unwrap(), live);
+        // Both changed `b`, differently: a conflict (nothing is applied). The same way: fine.
+        let mut both = live.clone();
+        set(&mut both, 3, 9.0);
+        assert!(merge_group(&base, &both, &result).is_none());
+        set(&mut both, 3, 2.0);
+        assert_eq!(get(&merge_group(&base, &both, &result).unwrap(), 3), 2.0);
+        // The analysis adds a child (a result group) while the user edits `a`: both kept.
+        let mut added = result.clone();
+        added.children.push(Node::Group(PropGroup::new(6, "out", "Output")));
+        let m = merge_group(&base, &live, &added).unwrap();
+        assert!(m.find_group(6).is_some() && get(&m, 2) == 1.0);
+        // The analysis drops the sub-group the user didn't touch: dropped.
+        let mut dropped = result.clone();
+        dropped.children.retain(|n| !matches!(n, Node::Group(g) if g.uid == 4));
+        let m = merge_group(&base, &live, &dropped).unwrap();
+        assert!(m.find_group(4).is_none() && get(&m, 2) == 1.0);
+        // …but if the user changed something in it, that's a conflict.
+        let mut touched = live.clone();
+        set(&mut touched, 5, 7.0);
+        assert!(merge_group(&base, &touched, &dropped).is_none());
+    }
+
     #[test]
     fn protocol_round_trips() {
         let mut s = Session::default();

crates/engine/src/tests_mask_warp.rs

@@ -412,6 +412,162 @@ impl Deferred {
     }
 }
 
+/// #1249: a setting changed while the analysis runs in a worker survives its result, which
+/// used to replace the whole effect with the copy the job started from.
+#[test]
+fn an_edit_made_during_an_offloaded_analysis_survives_its_result() {
+    let (mut s, clip, solid) = setup(shaky);
+    s.execute("edit.clear", json!({"layers": [solid.0]})).unwrap();
+    s.execute("layer.select", json!({"layers": [clip.0]})).unwrap();
+    s.execute("effect.apply", json!({"effect": "Warp Stabilizer"})).unwrap();
+    set_param(&mut s, clip, "stabilization/smoothness", json!(37.5));
+    let off = std::sync::Arc::new(Deferred::default());
+    s.offload = Some(off.clone());
+    s.execute_checked("warp.analyze", json!({"layer": clip.0})).unwrap();
+    // While the job runs: another setting of the same effect.
+    set_param(&mut s, clip, "stabilization/smoothness", json!(12.5));
+    let fx = |s: &Session| s.active_comp().unwrap().layer(clip).unwrap().props.prop("effects/#1/stabilization/smoothness").unwrap().value.clone();
+    off.run(s.footage.clone());
+    assert!(s.poll_offload());
+    assert!(!s.drain_events().iter().any(|e| matches!(e, crate::Event::Toast { error: true, .. })), "applied without a conflict");
+    let st = s.execute("warp.status", json!({})).unwrap();
+    assert_eq!(st["analyzed"], true, "the analysis landed: {st}");
+    assert_eq!(fx(&s), KV::Scalar(12.5), "the edit made during the analysis stays");
+    // One undo step takes the analysis back and keeps the edit.
+    s.execute("edit.undo", json!({})).unwrap();
+    assert_eq!(fx(&s), KV::Scalar(12.5));
+    assert_eq!(s.execute("warp.status", json!({})).unwrap()["analyzed"], false);
+}
+
+fn point_track_pose(f: u32) -> ([f64; 2], f64, f64) {
+    ([150.0 + 2.0 * f as f64, 110.0 + f as f64], 0.0, 1.0)
+}
+
+/// Hold an actual point-tracking worker request so commands can edit its target before the
+/// serialized worker reply is delivered.
+fn held_point_track() -> (Session, LayerId, Uid, std::sync::Arc<Deferred>) {
+    let (mut s, clip, solid) = setup(patch_frames(point_track_pose));
+    s.execute("edit.clear", json!({"layers": [solid.0]})).unwrap();
+    s.execute("layer.select", json!({"layers": [clip.0]})).unwrap();
+    let tracker = s.execute("track.motion", json!({})).unwrap()["tracker"].as_u64().unwrap();
+    s.execute("track.setPoint", json!({"point": 1, "center": point_track_pose(0).0, "featureSize": [36, 36], "searchSize": [80, 80]})).unwrap();
+    let off = std::sync::Arc::new(Deferred::default());
+    s.offload = Some(off.clone());
+    s.execute_checked("track.analyze", json!({"direction": "forward"})).unwrap();
+    assert_eq!(off.kinds(), [crate::offload::JobKind::Track]);
+    (s, clip, tracker, off)
+}
+
+fn tracked_center(s: &Session, clip: LayerId, tracker: Uid) -> Uid {
+    s.active_comp().unwrap().layer(clip).unwrap().props.find_group(tracker).unwrap().track_points().next().unwrap().get("featureCenter").unwrap().uid
+}
+
+#[test]
+fn offloaded_conflict_preserves_state_and_reports_completion_failure() {
+    let (mut s, clip, tracker, off) = held_point_track();
+    let center = tracked_center(&s, clip, tracker);
+    s.execute("prop.set", json!({"layer": clip.0, "prop": center, "value": [180.0, 140.0]})).unwrap();
+    s.execute("prop.renameGroup", json!({"layer": clip.0, "prop": tracker, "name": "Redo survives"})).unwrap();
+    assert!(s.undo());
+    let before = s.project.clone();
+    let state = serde_json::to_value(&s.state).unwrap();
+    let steps = |s: &Session| {
+        let entries =
+            |v: &[(String, std::sync::Arc<effectcraft_project::Project>)]| v.iter().map(|(label, p)| (label.clone(), p.to_json())).collect::<Vec<_>>();
+        (
+            entries(&s.history.undo),
+            entries(&s.history.redo),
+            s.history.branches.iter().map(|b| (b.parent.to_json(), entries(&b.steps))).collect::<Vec<_>>(),
+            s.history.merge_key.clone(),
+        )
+    };
+    let history = steps(&s);
+    let inbox = s.offloaded(crate::offload::JobKind::Track).unwrap().inbox.clone();
+    off.run(s.footage.clone());
+    let (done, replies): (Vec<_>, Vec<_>) = inbox.take().into_iter().partition(|r| matches!(r, crate::offload::WorkerReply::Done));
+    assert_eq!(done.len(), 1);
+    for reply in replies {
+        inbox.push(reply);
+    }
+    s.drain_events();
+    assert!(s.poll_offload());
+    let pending = s.offloaded(crate::offload::JobKind::Track).unwrap();
+    assert!(pending.progress.error.as_ref().is_some_and(|e| e.contains("nothing was applied")));
+    assert!(!pending.progress.finished);
+    assert!(s.offload_log.is_empty(), "the error is kept before completion is logged");
+    // The current completion remains identifiable after the bounded log evicts an old entry.
+    s.offload_log = vec![(crate::offload::JobKind::Warp, None); 32];
+    for reply in done {
+        inbox.push(reply);
+    }
+    assert!(s.poll_offload());
+    assert!(std::sync::Arc::ptr_eq(&s.project, &before), "a rejected result changes no document content");
+    assert_eq!(serde_json::to_value(&s.state).unwrap(), state);
+    assert_eq!(steps(&s), history);
+    let errors: Vec<_> = s
+        .drain_events()
+        .into_iter()
+        .filter_map(|e| match e {
+            crate::Event::Toast { error: true, message } => Some(message),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(errors.len(), 1, "the apply rejection is reported once: {errors:?}");
+    assert!(errors[0].contains("nothing was applied"));
+    assert_eq!(s.offload_log.len(), 32);
+    assert_eq!(s.offload_log_start, 1);
+    let (kind, error) = s.offload_log.last().unwrap();
+    assert_eq!(*kind, crate::offload::JobKind::Track);
+    assert!(error.as_ref().is_some_and(|e| e.contains("nothing was applied")), "a rejected result must report failure to JobWaiter");
+    assert!(s.redo());
+    assert_eq!(s.active_comp().unwrap().layer(clip).unwrap().props.find_group(tracker).unwrap().name, "Redo survives");
+}
+
+#[test]
+fn offloaded_disjoint_edit_reports_success_and_keeps_one_undo_step() {
+    let (mut s, clip, tracker, off) = held_point_track();
+    s.execute("prop.renameGroup", json!({"layer": clip.0, "prop": tracker, "name": "Live track"})).unwrap();
+    s.execute("prop.setGroupEnabled", json!({"layer": clip.0, "prop": tracker, "value": false})).unwrap();
+    let undo = s.history.undo.len();
+    off.run(s.footage.clone());
+    s.drain_events();
+    assert!(s.poll_offload());
+    let layer = s.active_comp().unwrap().layer(clip).unwrap();
+    let track = layer.props.find_group(tracker).unwrap();
+    assert_eq!(track.name, "Live track");
+    assert!(!track.enabled);
+    assert_eq!(track.track_points().next().unwrap().get("featureCenter").unwrap().keys.len(), 25);
+    assert_eq!(s.history.undo.len(), undo + 1);
+    assert_eq!(s.offload_log.last(), Some(&(crate::offload::JobKind::Track, None)));
+    assert!(!s.drain_events().iter().any(|e| matches!(e, crate::Event::Toast { error: true, .. })));
+    let applied = s.project.to_json();
+    assert!(s.undo());
+    let track = s.active_comp().unwrap().layer(clip).unwrap().props.find_group(tracker).unwrap();
+    assert_eq!(track.name, "Live track");
+    assert!(!track.enabled);
+    assert!(track.track_points().next().unwrap().get("featureCenter").unwrap().keys.is_empty());
+    assert!(s.redo());
+    assert_eq!(s.project.to_json(), applied);
+}
+
+#[test]
+fn offloaded_conflict_does_not_fail_a_concurrent_analysis() {
+    let (mut s, clip, tracker, off) = held_point_track();
+    let effect = s.execute("effect.apply", json!({"effect": "Warp Stabilizer", "layers": [clip.0]})).unwrap()["effects"][0].as_u64().unwrap();
+    s.execute_checked("warp.analyze", json!({"layer": clip.0, "effect": effect})).unwrap();
+    assert_eq!(off.kinds(), [crate::offload::JobKind::Track, crate::offload::JobKind::Warp]);
+    let center = tracked_center(&s, clip, tracker);
+    s.execute("prop.set", json!({"layer": clip.0, "prop": center, "value": [180.0, 140.0]})).unwrap();
+    off.run(s.footage.clone());
+    assert!(s.poll_offload());
+    let track = s.offload_log.iter().find(|(kind, _)| *kind == crate::offload::JobKind::Track).unwrap();
+    assert!(track.1.as_ref().is_some_and(|e| e.contains("nothing was applied")));
+    let warp = s.offload_log.iter().find(|(kind, _)| *kind == crate::offload::JobKind::Warp).unwrap();
+    assert!(warp.1.is_none(), "the other analysis completed successfully: {warp:?}");
+    assert_eq!(s.execute("warp.status", json!({"layer": clip.0, "effect": effect})).unwrap()["analyzed"], true);
+    assert!(s.active_comp().unwrap().layer(clip).unwrap().props.find(center).unwrap().keys.is_empty());
+}
+
 /// M13.32: where jobs run elsewhere (the browser), `warp.status` never solves the
 /// stabilization plan on the page's thread (seconds for a long Subspace Warp clip): the
 /// analysis job sends the plan's summary with its result, and a settings change has the plan

crates/engine/src/tests_settings.rs

@@ -548,6 +548,66 @@ fn frontend_commands_are_bindable_with_panel_scope() {
 
 // ---------------------------------------------------------------- auto-save / recovery
 
+/// The names in `d`, sorted.
+fn names(d: &std::path::Path) -> Vec<String> {
+    let mut v: Vec<String> = std::fs::read_dir(d).unwrap().map(|e| e.unwrap().file_name().to_string_lossy().into_owned()).collect();
+    v.sort();
+    v
+}
+
+#[cfg(unix)]
+#[test]
+fn atomic_write_overwrites_a_long_valid_filename() {
+    let d = tmp("atomic-long-name");
+    // The old `<name>.tmp` fits a 255-byte component limit for this 251-byte project name.
+    // Including the full name plus a process id and counter made an otherwise valid save fail.
+    let name = format!("{}.ecproj", "x".repeat(244));
+    assert_eq!(name.len(), 251);
+    let f = d.join(&name);
+    std::fs::write(&f, b"original project").unwrap();
+    let replacement = b"replacement project\0bytes\xff";
+    atomic_write(&f, replacement).unwrap();
+    assert_eq!(std::fs::read(&f).unwrap(), replacement);
+    assert_eq!(names(&d), [name], "no temporary file is left after overwriting the project");
+    let _ = std::fs::remove_dir_all(d);
+}
+
+#[test]
+fn concurrent_atomic_writes_each_publish_a_whole_file() {
+    // Writers of the same file used to share one `.tmp`: they truncated each other's data and
+    // the later rename failed once the first had moved it away (#1233).
+    let d = tmp("atomic-concurrent");
+    let f = d.join("shared.ecproj");
+    for round in 0..10 {
+        let contents: Vec<Vec<u8>> = (0..8u8).map(|w| vec![b'a' + w; 256 * 1024 + usize::from(w) * 1000]).collect();
+        std::thread::scope(|sc| {
+            for c in &contents {
+                let f = &f;
+                sc.spawn(move || atomic_write(f, c).expect("every writer succeeds"));
+            }
+        });
+        let got = std::fs::read(&f).unwrap();
+        assert!(contents.contains(&got), "round {round}: the file is one writer's whole data ({} bytes)", got.len());
+        assert_eq!(names(&d), ["shared.ecproj"], "round {round}: no temporary file is left");
+    }
+    let _ = std::fs::remove_dir_all(d);
+}
+
+#[cfg(unix)]
+#[test]
+fn atomic_write_does_not_follow_a_link_at_the_old_temp_name() {
+    let d = tmp("atomic-link");
+    let f = d.join("p.ecproj");
+    let victim = d.join("victim.txt");
+    std::fs::write(&victim, b"keep").unwrap();
+    // The temporary name used to be `<name>.tmp`, opened with File::create (following links).
+    std::os::unix::fs::symlink(&victim, d.join("p.ecproj.tmp")).unwrap();
+    atomic_write(&f, b"new project").unwrap();
+    assert_eq!(std::fs::read(&victim).unwrap(), b"keep", "the link's target is untouched");
+    assert_eq!(std::fs::read(&f).unwrap(), b"new project");
+    let _ = std::fs::remove_dir_all(d);
+}
+
 #[test]
 fn atomic_write_survives_a_crash_mid_write() {
     let d = tmp("atomic");
@@ -555,14 +615,12 @@ fn atomic_write_survives_a_crash_mid_write() {
     atomic_write(&f, b"version 1").unwrap();
     // The writer dies halfway through the temporary file.
     let r = atomic_write_with(&f, b"version 2 is much longer", |tmp, data| {
-        std::fs::write(tmp, &data[..5])?;
+        std::io::Write::write_all(tmp, &data[..5])?;
         Err(std::io::Error::other("power cut"))
     });
     assert!(r.is_err());
     assert_eq!(std::fs::read(&f).unwrap(), b"version 1");
-    let leftovers: Vec<String> =
-        std::fs::read_dir(&d).unwrap().flatten().map(|e| e.file_name().to_string_lossy().to_string()).filter(|n| n != "p.ecproj").collect();
-    assert!(leftovers.is_empty(), "the partial temporary is cleaned up: {leftovers:?}");
+    assert_eq!(names(&d), ["p.ecproj"], "the partial temp file is cleaned up");
     // A temp file left by a hard kill doesn't count as an auto-save.
     std::fs::write(d.join("p auto-save 1.ecproj.tmp"), b"junk").unwrap();
     assert!(autosave::existing(&d, "p").is_empty());
@@ -571,22 +629,11 @@ fn atomic_write_survives_a_crash_mid_write() {
     let _ = std::fs::remove_dir_all(d);
 }
 
-#[cfg(unix)]
-#[test]
-fn atomic_write_does_not_follow_a_planted_temporary_symlink() {
-    let d = tmp("atomic-link");
-    let f = d.join("p.ecproj");
-    let sentinel = d.join("sentinel.txt");
-    std::fs::write(&sentinel, b"precious").unwrap();
-    // Someone plants a symlink at the temporary name the write used to go through.
-    std::os::unix::fs::symlink(&sentinel, d.join("p.ecproj.tmp")).unwrap();
-    atomic_write(&f, b"new").unwrap();
-    assert_eq!(std::fs::read(&sentinel).unwrap(), b"precious", "a symlink at the temporary name is never followed");
-    assert_eq!(std::fs::read(&f).unwrap(), b"new");
-    let _ = std::fs::remove_dir_all(d);
-}
-
-#[cfg(unix)]
+/// A writer that is paused mid-temporary while another writer fully publishes to the same
+/// destination must not let either one publish a mixture: the bounded, exclusively-created
+/// temporaries keep the two writes apart (#1233). Unlike
+/// `concurrent_atomic_writes_each_publish_a_whole_file`, this pins the interleaving
+/// deterministically instead of sampling it under real contention.
 #[test]
 fn atomic_writes_in_parallel_never_publish_a_mixture() {
     use std::sync::{Arc, Barrier};
@@ -594,15 +641,14 @@ fn atomic_writes_in_parallel_never_publish_a_mixture() {
     let f = d.join("p.ecproj");
     let a = vec![b'A'; 64];
     let b = vec![b'B'; 64];
-    // Writer A writes the first half of its temporary, then waits while writer B runs against
-    // the same destination, and only then finishes.
+    // Writer A writes the first half of its temporary, then waits while writer B fully publishes
+    // against the same destination, and only then finishes.
     let (half, published) = (Arc::new(Barrier::new(2)), Arc::new(Barrier::new(2)));
     let (f2, a2) = (f.clone(), a.clone());
     let (half2, published2) = (half.clone(), published.clone());
     let t = std::thread::spawn(move || {
-        atomic_write_with(&f2, &a2, |tmp, data| {
+        atomic_write_with(&f2, &a2, |fh, data| {
             use std::io::Write;
-            let mut fh = std::fs::File::create(tmp)?;
             fh.write_all(&data[..32])?;
             half2.wait();
             published2.wait();
@@ -613,7 +659,7 @@ fn atomic_writes_in_parallel_never_publish_a_mixture() {
     half.wait();
     atomic_write(&f, &b).unwrap();
     published.wait();
-    let _ = t.join();
+    assert!(t.join().unwrap().is_ok(), "the paused writer still publishes");
     let got = std::fs::read(&f).unwrap();
     assert!(got == a || got == b, "the destination holds one complete write, never a mixture: {}", String::from_utf8_lossy(&got));
     let _ = std::fs::remove_dir_all(d);

crates/pdf/src/ps.rs

@@ -700,20 +700,10 @@ impl Interp {
                             a / b
                         }
                     }
-                    "idiv" => {
-                        if b as i64 == 0 {
-                            0.0
-                        } else {
-                            ((a as i64) / (b as i64)) as f64
-                        }
-                    }
-                    "mod" => {
-                        if b as i64 == 0 {
-                            0.0
-                        } else {
-                            ((a as i64) % (b as i64)) as f64
-                        }
-                    }
+                    // `checked_*` give 0 both for a zero divisor and for the one overflowing
+                    // case, `i64::MIN / -1` (`/` and `%` panic on it in every build).
+                    "idiv" => (a as i64).checked_div(b as i64).unwrap_or(0) as f64,
+                    "mod" => (a as i64).checked_rem(b as i64).unwrap_or(0) as f64,
                     "atan" => a.atan2(b).to_degrees().rem_euclid(360.0),
                     "exp" => a.powf(b),
                     "max" => a.max(b),

crates/pdf/src/tests_import.rs

@@ -376,3 +376,25 @@ fn postscript_radix_numbers_outside_2_to_36_are_not_numbers() {
         let _ = parse(src.as_bytes());
     }
 }
+
+#[test]
+fn postscript_idiv_and_mod_of_i64_min_by_minus_one_give_zero() {
+    // `i64::MIN / -1` and `i64::MIN % -1` overflow, and `/` and `%` panic on them in debug and
+    // release builds. `as i64` turns any dividend at or below -2^63 into `i64::MIN` and turns
+    // a divisor of -1.5 into -1.
+    for (prog, want) in [
+        ("-9223372036854775808 -1 idiv", 0),
+        ("-9223372036854775808 -1 mod", 0),
+        ("-99999999999999999999 -1.5 idiv", 0),
+        ("2 63 exp neg -1 mod", 0),
+        // The result is 0 for a divisor that truncates to 0. Other quotients are truncated
+        // toward zero, and a remainder has the sign of the dividend.
+        ("7 0 idiv", 0),
+        ("7 0.5 mod", 0),
+        ("-7 2 idiv", -3),
+        ("-7 2 mod", -1),
+    ] {
+        let doc = eps(format!("{prog} {want} eq {{ 0 0 10 10 rectfill }} if\n").as_bytes());
+        assert_eq!(shape_names(&doc).len(), 1, "{prog}");
+    }
+}

crates/ui-egui/src/control.rs

@@ -68,8 +68,8 @@ pub struct JobWaiter {
     pub result: Value,
     /// Ids of the jobs it started (`jobs.list` ids: `render`, `track`, `roto`…).
     pub jobs: Vec<String>,
-    /// Length of [`effectcraft_engine::Session::offload_log`] when it started.
-    pub log_len: usize,
+    /// Absolute completion-log position when it started (including evicted entries).
+    pub log_cursor: usize,
 }
 
 /// Commands that block with `wait` (and its default).
@@ -86,7 +86,7 @@ fn execute_engine(app: &mut EffectcraftApp, id: &str, params: Value) -> Result<O
         return app.session.execute_checked(id, params).map(ok).map_err(|e| e.to_string());
     }
     let before: Vec<String> = app.session.jobs().into_iter().map(|j| j.id).collect();
-    let log_len = app.session.offload_log.len();
+    let log_cursor = app.session.offload_log_start.saturating_add(app.session.offload_log.len());
     let mut p = params.clone();
     p["wait"] = json!(false);
     let result = app.session.execute_checked(id, p).map_err(|e| e.to_string())?;
@@ -94,7 +94,7 @@ fn execute_engine(app: &mut EffectcraftApp, id: &str, params: Value) -> Result<O
     if jobs.is_empty() {
         return Ok(ok(result));
     }
-    Ok(Outcome::AwaitJobs(JobWaiter { command: id.to_string(), params, result, jobs, log_len }))
+    Ok(Outcome::AwaitJobs(JobWaiter { command: id.to_string(), params, result, jobs, log_cursor }))
 }
 
 impl JobWaiter {
@@ -108,8 +108,13 @@ impl JobWaiter {
     pub fn finish(self, app: &mut EffectcraftApp) -> Value {
         let s = &mut app.session;
         s.poll_render();
-        let errors: Vec<String> = s.offload_log.iter().skip(self.log_len).filter_map(|(_, e)| e.clone()).collect();
-        if let Some(e) = errors.first() {
+        let error = s
+            .offload_log
+            .iter()
+            .skip(self.log_cursor.saturating_sub(s.offload_log_start))
+            .filter(|(kind, _)| self.jobs.iter().any(|id| id == kind.job_id()))
+            .find_map(|(_, e)| e.as_ref());
+        if let Some(e) = error {
             return json!({"ok": false, "error": e});
         }
         let mut r = self.result;

crates/ui-egui/tests/ui_web_offload.rs

@@ -219,3 +219,45 @@ fn waiting_commands_reply_when_their_offloaded_job_ends() {
     let v = reply.try_recv().expect("immediate reply");
     assert_eq!(v["result"]["rendering"], true, "{v}");
 }
+
+/// Completion errors belong to the waiting command's jobs, including when several
+/// different analyses complete in the same UI frame.
+#[test]
+fn waiting_commands_report_only_their_own_completion_errors() {
+    use effectcraft_engine::offload::JobKind;
+    use effectcraft_ui_egui::control::JobWaiter;
+
+    for (kind, job_id) in [
+        (JobKind::Render, "render"),
+        (JobKind::Warp, "warp"),
+        (JobKind::WarpPlan, "warpPlan"),
+        (JobKind::Camera, "camera"),
+        (JobKind::RotoFreeze, "roto"),
+        (JobKind::RotoPropagate, "roto"),
+        (JobKind::Track, "track"),
+        (JobKind::MaskTrack, "maskTrack"),
+        (JobKind::ContentFill, "contentFill"),
+    ] {
+        let other = if kind == JobKind::Track { JobKind::Warp } else { JobKind::Track };
+        let mut app = EffectcraftApp::new(Session::default());
+        app.session.offload_log = vec![(kind, Some("old error".into())), (other, Some("unrelated failure".into())), (kind, None)];
+        let waiter = || JobWaiter { command: "analysis".into(), params: json!({}), result: json!({"running": true}), jobs: vec![job_id.into()], log_cursor: 1 };
+        let result = waiter().finish(&mut app);
+        assert_eq!(result, json!({"ok": true, "result": {"running": false}}), "{kind:?}");
+        app.session.offload_log.push((kind, Some("nothing was applied".into())));
+        let result = waiter().finish(&mut app);
+        assert_eq!(result, json!({"ok": false, "error": "nothing was applied"}), "{kind:?}");
+    }
+}
+
+#[test]
+fn waiting_command_sees_its_error_after_the_completion_log_rolls_over() {
+    use effectcraft_engine::offload::JobKind;
+    use effectcraft_ui_egui::control::JobWaiter;
+    let mut app = EffectcraftApp::new(Session::default());
+    app.session.offload_log = vec![(JobKind::Warp, None); 31];
+    app.session.offload_log.push((JobKind::Track, Some("nothing was applied".into())));
+    app.session.offload_log_start = 1;
+    let waiter = JobWaiter { command: "track.analyze".into(), params: json!({}), result: json!({}), jobs: vec!["track".into()], log_cursor: 32 };
+    assert_eq!(waiter.finish(&mut app), json!({"ok": false, "error": "nothing was applied"}));
+}

docs/agents.md

@@ -52,6 +52,11 @@ Desktop, Cursor and the like) take the same `command` and `args`.
 - **Bridge** (`["mcp", "--bridge", "9877"]`): drives a running `effectcraft --control 9877`, so you
   see every change live. Bridge mode adds `screenshot` and the `ui_*` tools.
 
+After a broken connection, the bridge retries explicitly read-only control queries once.
+Engine commands, UI edits and unknown methods whose transmission or reply fails report an
+unknown outcome: the app may already have applied them. Inspect the app before explicitly
+retrying. A later call reconnects; timeouts and malformed replies are never automatically resent.
+
 ### From an installed release
 
 The release packages ship `effectcraft-cli` alongside the desktop app, so no build is needed: