From 0311bdc34d0dc10c2d18ff7875b4f52d483e5e27 Mon Sep 17 00:00:00 2001 From: Moein Saharkhiz <142007707+onlymoein@users.noreply.github.com> Date: Thu, 15 Oct 2026 04:13:41 -0400 Subject: [PATCH] fix(automation): avoid replaying edits after lost replies Retry only explicitly read-only control queries after connection failures. Report ambiguous edit outcomes without resending mutations, reconnect on the next explicit call, and reject request-ID exhaustion without overflow. Add loopback regressions for lost and malformed replies, read-only retries, playback queries versus edits, and unknown methods. Fixes #1247 --- crates/automation/src/bridge.rs | 17 +++-- crates/automation/src/tests.rs | 122 ++++++++++++++++++++++++++++++++ docs/agents.md | 5 ++ 3 files changed, 139 insertions(+), 5 deletions(-) diff --git a/crates/automation/src/bridge.rs b/crates/automation/src/bridge.rs index eab1c01d..77d69dd0 100644 --- a/crates/automation/src/bridge.rs +++ b/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 { + 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> { 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) }); } } } diff --git a/crates/automation/src/tests.rs b/crates/automation/src/tests.rs index 6bb6fbc5..6e622b37 100644 --- a/crates/automation/src/tests.rs +++ b/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(); diff --git a/docs/agents.md b/docs/agents.md index 1bec71d5..ed0073d4 100644 --- a/docs/agents.md +++ b/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: