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
This commit is contained in:
Moein Saharkhiz
2026-10-15 02:13:41 -06:00
committed by GitHub
parent 26ed1a35bf
commit 0311bdc34d
3 changed files with 139 additions and 5 deletions
+12 -5
View File
@@ -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) });
}
}
}
+122
View File
@@ -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();
+5
View File
@@ -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: