shifathosn
/
photocraft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Do not replay bridge requests after an uncertain result (#1527)

This commit is contained in:
Harrison O'Connor-Hoover 2026-10-09 00:11:44 -04:00 committed by GitHub
parent 9b08da8d98
commit de9cc5f12c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 99 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
crates/automation/src/bridge.rs
View File

@ -1,6 +1,6 @@
//! Client for the desktop app's JSON-lines control protocol
//! (`docs/control-protocol.md`): one JSON request per line, replies matched
//! by `id`. Keeps one connection and reconnects on failure.
//! by `id`. Keeps one connection and reconnects for the next call after a failure.
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
@ -50,58 +50,55 @@ impl BridgeClient {
/// Call a control method; returns its `result` or the app's error.
pub async fn call(&self, method: &str, params: Value) -> Result<Value, AutomationError> {
let mut guard = self.conn.lock().await;
// One retry with a fresh connection (the app may have restarted).
for attempt in 0..2 {
if guard.is_none() {
let s = tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr))
.await
.map_err(|_| AutomationError::Bridge(format!("timed out connecting to {}", self.addr)))?
.map_err(|e| {
AutomationError::Bridge(format!(
"cannot connect to {} ({e}); start the app with `photocraft --control <port>` and matching control credentials",
self.addr
))
})?;
let (r, w) = s.into_split();
let mut conn = (BufReader::new(r), w);
let auth_id = self.next_id.fetch_add(1, Ordering::Relaxed);
let auth = tokio::time::timeout(Duration::from_secs(5), exchange(&mut conn, auth_id, AUTH_METHOD, &json!({"token": self.token})))
.await
.map_err(|_| AutomationError::Bridge("control authentication timed out".into()))??;
auth?;
*guard = Some(conn);
if guard.is_none() {
let s = tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr))
.await
.map_err(|_| AutomationError::Bridge(format!("timed out connecting to {}", self.addr)))?
.map_err(|e| {
AutomationError::Bridge(format!(
"cannot connect to {} ({e}); start the app with `photocraft --control <port>` and matching control credentials",
self.addr
))
})?;
let (r, w) = s.into_split();
let mut conn = (BufReader::new(r), w);
let auth_id = self.next_id.fetch_add(1, Ordering::Relaxed);
let auth = tokio::time::timeout(Duration::from_secs(5), exchange(&mut conn, auth_id, AUTH_METHOD, &json!({"token": self.token})))
.await
.map_err(|_| AutomationError::Bridge("control authentication timed out".into()))??;
auth?;
*guard = Some(conn);
}
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
let Some(conn) = guard.as_mut() else {
return Err(AutomationError::Bridge(format!("not connected to {}", self.addr)));
};
match tokio::time::timeout(self.timeout, exchange(conn, id, method, &params)).await {
Ok(Ok(Err(error @ AutomationError::BadRequest(_)))) => {
// An oversized frame leaves unread bytes. Drop this connection and
// report the budget failure without retrying a possibly completed edit.
*guard = None;
Err(error)
}
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
let Some(conn) = guard.as_mut() else {
return Err(AutomationError::Bridge(format!("not connected to {}", self.addr)));
};
match tokio::time::timeout(self.timeout, exchange(conn, id, method, &params)).await {
Ok(Ok(Err(error @ AutomationError::BadRequest(_)))) => {
// An oversized frame leaves unread bytes. Drop this connection and
// report the budget failure without retrying a possibly completed edit.
*guard = None;
return Err(error);
}
Ok(Ok(v)) => return v,
Ok(Err(e)) if attempt == 0 => {
*guard = None;
let _ = e;
}
Ok(Err(e)) => {
*guard = None;
return Err(e);
}
Err(_) => {
*guard = None;
return Err(AutomationError::Bridge(format!("`{method}` timed out after {:?}", self.timeout)));
}
Ok(Ok(v)) => v,
Ok(Err(e)) => {
// A failed write or lost reply does not prove the edit was not applied.
// Reconnect on the next call, but never resend this operation (#1007).
*guard = None;
Err(AutomationError::Bridge(format!("`{method}` failed: {e}; operation may have completed; inspect state before retrying")))
}
Err(_) => {
*guard = None;
Err(AutomationError::Bridge(format!(
"`{method}` timed out after {:?}; operation may have completed; inspect state before retrying",
self.timeout
)))
}
}
Err(AutomationError::Bridge("unreachable".into()))
}
}
/// Outer `Err` = transport failure (retryable); inner = app-level result.
/// Outer `Err` = transport failure with an uncertain outcome; inner = app-level result.
async fn exchange(conn: &mut Conn, id: u64, method: &str, params: &Value) -> Result<Result<Value, AutomationError>, AutomationError> {
let mut line = serde_json::to_string(&json!({"id": id, "method": method, "params": params})).map_err(|e| AutomationError::Other(e.to_string()))?;
line.push('\n');
@ -115,7 +112,7 @@ async fn exchange(conn: &mut Conn, id: u64, method: &str, params: &Value) -> Res
loop {
buf.clear();
// Read raw bytes so truncation inside a UTF-8 character still reports the
// budget error instead of a retryable decoding/transport error.
// budget error instead of a decoding/transport error.
let n = (&mut conn.0).take((MAX_RESPONSE_BYTES + 1) as u64).read_until(b'\n', &mut buf).await.map_err(io)?;
if n > MAX_RESPONSE_BYTES {
return Ok(Err(AutomationError::BadRequest(format!("bridge response exceeds {MAX_RESPONSE_BYTES} bytes; operation may have completed"))));

53
crates/automation/tests/mcp.rs
View File

@ -507,6 +507,59 @@ async fn bridge_forwards_to_control_protocol() {
app.abort();
}
#[tokio::test(flavor = "multi_thread")]
async fn bridge_lost_reply_does_not_replay_an_edit_and_next_call_reconnects() {
use photocraft_automation::{BridgeClient, Headless};
use std::time::Duration;
tokio::time::timeout(Duration::from_secs(5), async {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap().to_string();
let app = tokio::spawn(async move {
let mut backend = Headless::new();
backend.handle("doc.new", json!({"width": 8, "height": 8})).unwrap();
let initial_layers = backend.session.active().unwrap().doc.layer_count();
let initial_history = backend.session.active().unwrap().history.past_len();
let mut seen = Vec::new();
for connection in 0..2 {
let (socket, _) = listener.accept().await.unwrap();
let (read, mut write) = socket.into_split();
let mut reader = BufReader::new(read);
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
let (auth, authenticated) = photocraft_automation::security::authentication_reply(&line, CONTROL_TOKEN);
assert!(authenticated, "each connection must authenticate");
write.write_all(format!("{auth}\n").as_bytes()).await.unwrap();
line.clear();
reader.read_line(&mut line).await.unwrap();
let request: Value = serde_json::from_str(&line).unwrap();
seen.push(request["params"]["command"].as_str().unwrap().to_owned());
let result = backend.handle(request["method"].as_str().unwrap(), request["params"].clone()).unwrap();
if connection == 0 {
// The edit has completed, but both socket halves close before its reply.
continue;
}
let reply = json!({"id": request["id"], "ok": true, "result": result});
write.write_all(format!("{reply}\n").as_bytes()).await.unwrap();
}
let document = backend.session.active().unwrap();
assert_eq!(document.doc.layer_count(), initial_layers + 1, "the edit was applied once");
assert_eq!(document.history.past_len(), initial_history + 1, "one undo step");
seen
});
let bridge = BridgeClient::new(&addr, CONTROL_TOKEN).unwrap();
let result = bridge.call("engine.execute", json!({"command": "layer.new.layer", "params": {"name": "Once"}})).await;
let error = result.unwrap_err().to_string();
assert!(error.contains("operation may have completed"), "{error}");
assert!(error.contains("inspect"), "{error}");
let document = bridge.call("engine.execute", json!({"command": "document.inspect"})).await.unwrap();
assert_eq!(document["layers"].as_array().unwrap().len(), 2);
assert_eq!(app.await.unwrap(), ["layer.new.layer", "document.inspect"]);
})
.await
.expect("bridge lost-reply regression must finish");
}
#[tokio::test(flavor = "multi_thread")]
async fn bridge_previews_downscale_before_enforcing_the_png_budget() {
use photocraft_automation::budgets::MAX_PNG_BYTES;

2
docs/control-protocol.md
View File

@ -170,7 +170,7 @@ magenta alignment lines. `ui.pointer` drives the same code, so agents get identi
- **Headless** (`photocraft-cli mcp`): an in-process `photocraft_engine::Session`. There is no window.
- **Bridge** (`photocraft-cli mcp --bridge 127.0.0.1:7878 --control-token-file <path>`): every tool is forwarded to a running `photocraft --control 7878 --control-token-file <path>` over this protocol, so agents see and drive the live app.
The bridge keeps one authenticated TCP connection open. It reconnects and authenticates once if a request fails, and it skips reply lines whose `id` doesn't match the request (for example, stale replies to requests that timed out). It only accepts loopback addresses, because the app only listens on loopback. Supply its bearer token with `--control-token-file`, `--control-token`, `PHOTOCRAFT_CONTROL_TOKEN_FILE`, or `PHOTOCRAFT_CONTROL_TOKEN`:
The bridge keeps one authenticated TCP connection open. A transport failure while sending a request or waiting for its reply, including a reply timeout, drops the connection and reports that the operation may have completed; inspect the document before retrying an edit. It never automatically resends the failed request. The next separate call reconnects and authenticates, and reply lines whose `id` does not match the request are skipped. It only accepts loopback addresses, because the app only listens on loopback. Supply its bearer token with `--control-token-file`, `--control-token`, `PHOTOCRAFT_CONTROL_TOKEN_FILE`, or `PHOTOCRAFT_CONTROL_TOKEN`:
```sh
photocraft-cli mcp --bridge 127.0.0.1:7878 \