mirror of
https://github.com/storytold/photocraft.git
synced 2026-10-14 19:26:51 -04:00
* MCP bridge: reconnect before idle sockets die, retry reads, actionable timeouts (#2487) The app closes idle control connections after 30s (IO_TIMEOUT) while the bridge kept one persistent connection, so the first call after a pause failed with "operation may have completed". The bridge now drops and reconnects before sending once the connection has been idle 25s, so a request is never written to a socket the app may have dropped. The default call timeout drops from 60s to 55s, strictly below the app's 60s reply deadline, so the bridge's own error — naming the method and the likely causes — is the one callers see instead of a race between the app's deadline reply and the caller's own cutoff. Idempotent read methods (ui.inspect and friends) get one automatic retry on a transport failure; edits keep #1007 semantics and are never resent. A timed-out call is never retried (it would only double the wait). * MCP bridge: caller-vouched read retries, per-failure wording, docs contract (#2487 review) The retry decision moves from a hand-maintained method list — most of whose entries were engine command ids that never travel as control-protocol methods, so they could never match — to the call sites that know a call is a read: call never resends (#1007), call_read retries once on a transport failure, and the MCP server opts its pure-read tools in. Failures are worded by what they prove: a connection that could never be established is "the request was not sent", not "may have completed"; a read timeout says a read may be retried safely instead of demanding a state inspection. The 55s client cap, the read-retry exception, and the idle reconnect are now documented in the control protocol and the security book.
This commit is contained in:
1 parent
ecbd3d78be
commit
3e3e580217
5 files changed
+321
-35
No files matched your search
@@ -19,7 +19,9 @@ Automation is a privilege boundary because requests can cause filesystem access,
|
||||
PNGs are limited to 5 MiB before base64 encoding or file writes. Trusted-local CLI rendering
|
||||
retains its existing behavior.
|
||||
- The MCP bridge bounds incoming replies and does not retry an operation after an oversized
|
||||
reply. Screenshots are decoded under separate dimension, pixel, and allocation ceilings before
|
||||
reply. Edits are never automatically resent after a transport failure (#1007); the server's
|
||||
pure-read tools opt into a single retry on a fresh connection, which applies nothing.
|
||||
Screenshots are decoded under separate dimension, pixel, and allocation ceilings before
|
||||
downscaling; the 5 MiB PNG limit applies to the returned image.
|
||||
- MCP normally uses stdio and can optionally bridge to loopback control TCP.
|
||||
- The bridge and UI request paths use timeouts.
|
||||
|
||||
+141
-23
@@ -1,9 +1,12 @@
|
||||
//! 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 for the next call after a failure.
|
||||
//! by `id`. Keeps one connection, reconnecting after a failure — or before
|
||||
//! a call, once the connection has been idle long enough that the app
|
||||
//! (which closes idle control connections after
|
||||
//! [`IO_TIMEOUT`](crate::security::IO_TIMEOUT)) may already have closed it.
|
||||
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde_json::{Value, json};
|
||||
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
|
||||
@@ -12,16 +15,31 @@ use tokio::sync::Mutex;
|
||||
|
||||
use crate::AutomationError;
|
||||
use crate::budgets::MAX_RESPONSE_BYTES;
|
||||
use crate::security::{AUTH_METHOD, MAX_REQUEST_BYTES, validate_token};
|
||||
use crate::security::{AUTH_METHOD, IO_TIMEOUT, MAX_REQUEST_BYTES, validate_token};
|
||||
|
||||
type Conn = (BufReader<tokio::net::tcp::OwnedReadHalf>, tokio::net::tcp::OwnedWriteHalf);
|
||||
|
||||
/// Reconnect before sending when the cached connection has been idle this long: the app
|
||||
/// drops idle control connections after [`IO_TIMEOUT`], and reconnecting before the send
|
||||
/// keeps the request out of the "was it applied?" state. The margin covers a request
|
||||
/// still in flight when the idle clock runs out.
|
||||
const IDLE_RECONNECT: Duration = IO_TIMEOUT.saturating_sub(Duration::from_secs(5));
|
||||
|
||||
/// The default call timeout sits strictly below the app's 60s reply deadline, so this
|
||||
/// client's error — not the app's deadline reply, not the caller's own cutoff — is the
|
||||
/// one the caller sees.
|
||||
pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(55);
|
||||
|
||||
const NO_REPLY_CAUSES: &str =
|
||||
"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)";
|
||||
|
||||
pub struct BridgeClient {
|
||||
addr: String,
|
||||
token: String,
|
||||
conn: Mutex<Option<Conn>>,
|
||||
conn: Mutex<Option<(Conn, Instant)>>,
|
||||
next_id: AtomicU64,
|
||||
timeout: Duration,
|
||||
idle_reconnect: Duration,
|
||||
}
|
||||
|
||||
impl BridgeClient {
|
||||
@@ -35,7 +53,14 @@ impl BridgeClient {
|
||||
return Err(AutomationError::BadRequest(format!("bridge address must be loopback, got `{addr}`")));
|
||||
}
|
||||
validate_token(&token)?;
|
||||
Ok(BridgeClient { addr, token: token.to_ascii_lowercase(), conn: Mutex::new(None), next_id: AtomicU64::new(1), timeout: Duration::from_secs(60) })
|
||||
Ok(BridgeClient {
|
||||
addr,
|
||||
token: token.to_ascii_lowercase(),
|
||||
conn: Mutex::new(None),
|
||||
next_id: AtomicU64::new(1),
|
||||
timeout: DEFAULT_CALL_TIMEOUT,
|
||||
idle_reconnect: IDLE_RECONNECT,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn with_timeout(mut self, t: Duration) -> Self {
|
||||
@@ -43,19 +68,69 @@ impl BridgeClient {
|
||||
self
|
||||
}
|
||||
|
||||
/// Shrink the idle-reconnect threshold; tests use it to run the reconnect path quickly.
|
||||
pub fn with_idle_reconnect(mut self, t: Duration) -> Self {
|
||||
self.idle_reconnect = t;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn addr(&self) -> &str {
|
||||
&self.addr
|
||||
}
|
||||
|
||||
/// Call a control method; returns its `result` or the app's error.
|
||||
/// Call a control method; returns its `result` or the app's error. Never resends after
|
||||
/// a transport failure: the request may have been applied (#1007).
|
||||
pub async fn call(&self, method: &str, params: Value) -> Result<Value, AutomationError> {
|
||||
self.dispatch(method, ¶ms, false).await
|
||||
}
|
||||
|
||||
/// [`Self::call`] for a method the caller knows only reads state. A transport failure
|
||||
/// applied nothing, so one fresh attempt is made on a new connection.
|
||||
pub async fn call_read(&self, method: &str, params: Value) -> Result<Value, AutomationError> {
|
||||
self.dispatch(method, ¶ms, true).await
|
||||
}
|
||||
|
||||
async fn dispatch(&self, method: &str, params: &Value, read: bool) -> Result<Value, AutomationError> {
|
||||
match self.attempt(method, params).await {
|
||||
Ok(v) => Ok(v),
|
||||
// A dropped socket or an unsent request applied nothing: safe to retry once,
|
||||
// and only for a call the caller vouched is a read.
|
||||
Err(Failure::NotSent(_)) | Err(Failure::Transport(_)) if read => match self.attempt(method, params).await {
|
||||
Ok(v) => Ok(v),
|
||||
Err(retry) => Err(self.failure(method, retry, read)),
|
||||
},
|
||||
Err(f) => Err(self.failure(method, f, read)),
|
||||
}
|
||||
}
|
||||
|
||||
/// The message for a failed attempt, worded by what the failure proves.
|
||||
fn failure(&self, method: &str, f: Failure, read: bool) -> AutomationError {
|
||||
match f {
|
||||
Failure::NotSent(e) => AutomationError::Bridge(format!("`{method}` failed: {e}; the request was not sent")),
|
||||
Failure::Transport(e) => AutomationError::Bridge(format!("`{method}` failed: {e}; operation may have completed; inspect state before retrying")),
|
||||
Failure::TimedOut(_) => {
|
||||
let what = if read { "a read may be retried safely" } else { "operation may have completed; inspect state before retrying" };
|
||||
AutomationError::Bridge(format!("`{method}` timed out after {:?}; {NO_REPLY_CAUSES}; {what}", self.timeout))
|
||||
}
|
||||
Failure::Definite(e) => e,
|
||||
}
|
||||
}
|
||||
|
||||
/// One exchange over one connection, reconnecting first when the cached one may be dead.
|
||||
async fn attempt(&self, method: &str, params: &Value) -> Result<Value, Failure> {
|
||||
let mut guard = self.conn.lock().await;
|
||||
if let Some((_, last_used)) = guard.as_ref()
|
||||
&& last_used.elapsed() >= self.idle_reconnect
|
||||
{
|
||||
// The request below goes to a fresh socket, not one the app may have dropped.
|
||||
*guard = None;
|
||||
}
|
||||
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(|_| Failure::NotSent(format!("timed out connecting to {}", self.addr)))?
|
||||
.map_err(|e| {
|
||||
AutomationError::Bridge(format!(
|
||||
Failure::NotSent(format!(
|
||||
"cannot connect to {} ({e}); start the app with `photocraft --control <port>` and matching control credentials",
|
||||
self.addr
|
||||
))
|
||||
@@ -65,39 +140,65 @@ impl BridgeClient {
|
||||
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);
|
||||
.map_err(|_| Failure::NotSent("control authentication timed out".into()))?
|
||||
.map_err(|e| Failure::NotSent(e.to_string()))?;
|
||||
if let Err(e) = auth {
|
||||
return Err(Failure::Definite(e));
|
||||
}
|
||||
*guard = Some((conn, Instant::now()));
|
||||
}
|
||||
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)));
|
||||
let Some((conn, _)) = guard.as_mut() else {
|
||||
return Err(Failure::NotSent(format!("not connected to {}", self.addr)));
|
||||
};
|
||||
match tokio::time::timeout(self.timeout, exchange(conn, id, method, ¶ms)).await {
|
||||
let outcome = tokio::time::timeout(self.timeout, exchange(conn, id, method, params)).await;
|
||||
match outcome {
|
||||
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)
|
||||
Err(Failure::Definite(error))
|
||||
}
|
||||
Ok(Ok(v)) => v,
|
||||
Ok(Ok(Ok(v))) => {
|
||||
if let Some((_, last_used)) = guard.as_mut() {
|
||||
*last_used = Instant::now();
|
||||
}
|
||||
Ok(v)
|
||||
}
|
||||
Ok(Ok(Err(e))) => {
|
||||
// The app answered: the operation did not apply — except the app's own
|
||||
// deadline reply, which fires while the command may still be running.
|
||||
if let Some((_, last_used)) = guard.as_mut() {
|
||||
*last_used = Instant::now();
|
||||
}
|
||||
Err(Failure::Definite(e))
|
||||
}
|
||||
// A failed write or lost reply does not prove the edit was not applied.
|
||||
// Reconnect on the next call; only a vouched read may resend (#1007).
|
||||
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(Failure::Transport(e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
*guard = None;
|
||||
Err(AutomationError::Bridge(format!(
|
||||
"`{method}` timed out after {:?}; operation may have completed; inspect state before retrying",
|
||||
self.timeout
|
||||
)))
|
||||
Err(Failure::TimedOut(()))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// How a single attempt failed, deciding the retry and the wording above.
|
||||
enum Failure {
|
||||
/// The connection could not be established or authenticated: nothing was sent.
|
||||
NotSent(String),
|
||||
/// The request was sent and the reply was lost: the outcome is uncertain.
|
||||
Transport(String),
|
||||
/// The call's own timeout elapsed. Retrying would only double the wait.
|
||||
TimedOut(()),
|
||||
/// The app answered (ok or error) or the request was refused before it was sent.
|
||||
Definite(AutomationError),
|
||||
}
|
||||
|
||||
/// 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()))?;
|
||||
@@ -133,3 +234,20 @@ async fn exchange(conn: &mut Conn, id: u64, method: &str, params: &Value) -> Res
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn the_call_timeout_stays_below_the_app_reply_deadline() {
|
||||
// At or above the app's 60s deadline, callers would see their own cutoff first.
|
||||
assert!(DEFAULT_CALL_TIMEOUT < Duration::from_secs(60));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_reconnect_stays_below_the_app_idle_cutoff() {
|
||||
// Reconnecting later than IO_TIMEOUT would send into a socket the app dropped.
|
||||
assert!(IDLE_RECONNECT < IO_TIMEOUT);
|
||||
}
|
||||
}
|
||||
@@ -436,7 +436,7 @@ impl PhotocraftMcp {
|
||||
if max_side.is_some_and(|side| side > MAX_PREVIEW_SIDE) {
|
||||
return Ok(fail(format!("automation preview side exceeds {MAX_PREVIEW_SIDE} pixels")));
|
||||
}
|
||||
let response = match b.call("ui.screenshot", json!({})).await {
|
||||
let response = match b.call_read("ui.screenshot", json!({})).await {
|
||||
Ok(response) => response,
|
||||
Err(error) => return Ok(fail(error)),
|
||||
};
|
||||
@@ -504,7 +504,7 @@ impl PhotocraftMcp {
|
||||
let Some(b) = self.bridge_client() else {
|
||||
return Ok(no_backend());
|
||||
};
|
||||
to_result(b.call("ui.inspect", json!({})).await)
|
||||
to_result(b.call_read("ui.inspect", json!({})).await)
|
||||
}
|
||||
|
||||
#[tool(description = "Open an image or document file and make it active. Returns index, size and import warnings.")]
|
||||
@@ -575,7 +575,7 @@ impl PhotocraftMcp {
|
||||
return Ok(no_backend());
|
||||
};
|
||||
let params = p.index.map(|index| json!({"document": index})).unwrap_or_else(|| json!({}));
|
||||
to_result(b.call("engine.execute", json!({"command": "document.inspect", "params": params})).await)
|
||||
to_result(b.call_read("engine.execute", json!({"command": "document.inspect", "params": params})).await)
|
||||
}
|
||||
|
||||
#[tool(description = "Render the flattened document and return it as a PNG image (bridge mode: window screenshot).")]
|
||||
@@ -625,7 +625,7 @@ impl PhotocraftMcp {
|
||||
let Some(b) = self.bridge_client() else {
|
||||
return Ok(no_backend());
|
||||
};
|
||||
b.call("engine.commands", json!({})).await
|
||||
b.call_read("engine.commands", json!({})).await
|
||||
};
|
||||
let all = match all {
|
||||
Ok(v) => v,
|
||||
@@ -657,13 +657,13 @@ impl PhotocraftMcp {
|
||||
#[tool(description = "List background jobs: running ones with progress (0-1), message and elapsed time, then the last few \
|
||||
that ended (state done|failed|cancelled with their result or error). Finished jobs are applied first.")]
|
||||
async fn jobs_list(&self) -> Result<CallToolResult, McpError> {
|
||||
self.jobs_call("jobs.list", json!({})).await
|
||||
self.jobs_call("jobs.list", json!({}), true).await
|
||||
}
|
||||
|
||||
#[tool(description = "Cancel a background job by id (or every running job when `job` is omitted). The document is left \
|
||||
unchanged.")]
|
||||
async fn jobs_cancel(&self, Parameters(p): Parameters<JobCancelParams>) -> Result<CallToolResult, McpError> {
|
||||
self.jobs_call("jobs.cancel", p.job.map_or_else(|| json!({}), |j| json!({"job": j}))).await
|
||||
self.jobs_call("jobs.cancel", p.job.map_or_else(|| json!({}), |j| json!({"job": j})), false).await
|
||||
}
|
||||
|
||||
#[tool(description = "Run several engine commands in one call (fewer round trips). Returns {completed, failed, \
|
||||
@@ -716,7 +716,7 @@ impl PhotocraftMcp {
|
||||
)]
|
||||
async fn ui_inspect(&self) -> Result<CallToolResult, McpError> {
|
||||
match self.bridge_client() {
|
||||
Some(b) => to_result(b.call("ui.inspect", json!({})).await),
|
||||
Some(b) => to_result(b.call_read("ui.inspect", json!({})).await),
|
||||
None => bridge_only("ui_inspect"),
|
||||
}
|
||||
}
|
||||
@@ -791,8 +791,8 @@ impl PhotocraftMcp {
|
||||
to_result(b.call("engine.execute", json!({"command": id, "params": params, "wait": wait})).await)
|
||||
}
|
||||
|
||||
/// `jobs.list` / `jobs.cancel` in either backend.
|
||||
async fn jobs_call(&self, method: &'static str, params: Value) -> Result<CallToolResult, McpError> {
|
||||
/// `jobs.list` / `jobs.cancel` in either backend; `read` picks the bridge's retry rule.
|
||||
async fn jobs_call(&self, method: &'static str, params: Value, read: bool) -> Result<CallToolResult, McpError> {
|
||||
let p2 = params.clone();
|
||||
if let Some(r) = self.headless_op(move |h| h.command_start(method, p2, true)).await {
|
||||
return to_result(r);
|
||||
@@ -800,7 +800,7 @@ impl PhotocraftMcp {
|
||||
let Some(b) = self.bridge_client() else {
|
||||
return Ok(no_backend());
|
||||
};
|
||||
to_result(b.call(method, params).await)
|
||||
if read { to_result(b.call_read(method, params).await) } else { to_result(b.call(method, params).await) }
|
||||
}
|
||||
|
||||
async fn save_impl(&self, p: SaveParams) -> Result<CallToolResult, McpError> {
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
//! Bridge client behavior against a scripted fake control server: reconnect before
|
||||
//! idle sockets die, retry only caller-vouched reads, never resend edits (#1007),
|
||||
//! never retry timeouts.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use photocraft_automation::security::authentication_reply;
|
||||
use photocraft_automation::{AutomationError, BridgeClient};
|
||||
use serde_json::{Value, json};
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
|
||||
const CONTROL_TOKEN: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
|
||||
|
||||
/// What the fake server does after serving one request on a connection.
|
||||
#[derive(Clone, Copy, PartialEq)]
|
||||
enum After {
|
||||
KeepOpen,
|
||||
Close,
|
||||
}
|
||||
|
||||
/// Serve connections until `max_connections` were seen. Each connection authenticates,
|
||||
/// then serves requests until it closes; every served request is recorded, and `accepted`
|
||||
/// counts connections as they arrive.
|
||||
async fn fake_server(
|
||||
max_connections: usize,
|
||||
after: After,
|
||||
seen: tokio::sync::mpsc::UnboundedSender<Vec<Value>>,
|
||||
accepted: std::sync::Arc<std::sync::atomic::AtomicUsize>,
|
||||
) -> String {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap().to_string();
|
||||
tokio::spawn(async move {
|
||||
let mut connections = 0;
|
||||
while connections < max_connections {
|
||||
let Ok((socket, _)) = listener.accept().await else { break };
|
||||
connections += 1;
|
||||
accepted.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
let seen = seen.clone();
|
||||
let after = after;
|
||||
tokio::spawn(async move {
|
||||
let mut served = Vec::new();
|
||||
let mut socket = socket;
|
||||
authenticate(&mut socket).await;
|
||||
let (read, mut write) = socket.into_split();
|
||||
let mut reader = BufReader::new(read);
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
line.clear();
|
||||
match reader.read_line(&mut line).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(_) => {}
|
||||
}
|
||||
let Ok(request) = serde_json::from_str::<Value>(&line) else { continue };
|
||||
served.push(request.clone());
|
||||
let reply = json!({"id": request["id"], "ok": true, "result": {"served": served.len()}});
|
||||
if write.write_all(format!("{reply}\n").as_bytes()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
if after == After::Close {
|
||||
break; // the reply is flushed, then both halves drop: a dead connection
|
||||
}
|
||||
}
|
||||
let _ = seen.send(served);
|
||||
});
|
||||
}
|
||||
});
|
||||
addr
|
||||
}
|
||||
|
||||
/// Authenticate one connection: read the auth line, reply to it.
|
||||
async fn authenticate(socket: &mut TcpStream) {
|
||||
let (read, mut write) = socket.split();
|
||||
let mut reader = BufReader::new(read);
|
||||
let mut line = String::new();
|
||||
reader.read_line(&mut line).await.unwrap();
|
||||
let (auth, ok) = authentication_reply(&line, CONTROL_TOKEN);
|
||||
assert!(ok, "each connection must authenticate");
|
||||
write.write_all(format!("{auth}\n").as_bytes()).await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn read_retries_on_a_dropped_connection_and_edits_do_not() {
|
||||
// Connection 1 serves one read then dies; the next vouched read retries on
|
||||
// connection 2. The edit, sent after connection 2 also died, must fail with the
|
||||
// #1007 wording and never open connection 3.
|
||||
let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
let accepted = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let addr = fake_server(2, After::Close, seen_tx, accepted).await;
|
||||
let bridge = BridgeClient::new(&addr, CONTROL_TOKEN).unwrap();
|
||||
|
||||
let first = bridge.call_read("document.inspect", json!({})).await.expect("read served before the connection dies");
|
||||
assert_eq!(first["served"], 1);
|
||||
|
||||
let retried = bridge.call_read("document.inspect", json!({})).await.expect("a vouched read retried on a fresh connection");
|
||||
assert_eq!(retried["served"], 1, "connection 2 sees it as its own first request");
|
||||
|
||||
let edit = bridge.call("engine.execute", json!({"command": "layer.new.layer", "params": {}})).await;
|
||||
let error = edit.unwrap_err().to_string();
|
||||
assert!(error.contains("operation may have completed"), "{error}");
|
||||
assert!(error.contains("inspect state before retrying"), "{error}");
|
||||
assert!(!error.contains("retried"), "{error}");
|
||||
|
||||
let mut all = Vec::new();
|
||||
while let Ok(served) = seen_rx.try_recv() {
|
||||
all.extend(served);
|
||||
}
|
||||
assert_eq!(all.len(), 2, "one request per connection, no edit resend: {all:?}");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_idle_connection_is_replaced_before_the_next_call() {
|
||||
// The production idle threshold sits under the app's 30s idle cutoff; the test
|
||||
// shrinks it so the reconnect path runs quickly. The next call after idling must
|
||||
// authenticate a new connection before sending.
|
||||
let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
let accepted = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let addr = fake_server(2, After::KeepOpen, seen_tx, std::sync::Arc::clone(&accepted)).await;
|
||||
let bridge = BridgeClient::new(&addr, CONTROL_TOKEN).unwrap().with_idle_reconnect(Duration::from_millis(100));
|
||||
|
||||
bridge.call_read("document.inspect", json!({})).await.expect("first call connects");
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
bridge.call_read("document.inspect", json!({})).await.expect("second call reconnects after idling");
|
||||
|
||||
assert_eq!(accepted.load(std::sync::atomic::Ordering::SeqCst), 2, "the idle connection was replaced, not reused");
|
||||
let first = seen_rx.try_recv().expect("the replaced connection closed, reporting what it served");
|
||||
assert_eq!(first.len(), 1, "each connection served only the request that followed its auth: {first:?}");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_timed_out_call_is_not_retried() {
|
||||
// The server authenticates but never answers requests, keeping the connection open:
|
||||
// the call hits the client's own timeout, which must name the likely causes and must
|
||||
// not double the wait by retrying (the fake server counts every request it sees).
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap().to_string();
|
||||
let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::<Value>::new()));
|
||||
let seen = std::sync::Arc::clone(&requests);
|
||||
tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
authenticate(&mut socket).await;
|
||||
let (read, _write) = socket.into_split();
|
||||
let mut reader = BufReader::new(read);
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
line.clear();
|
||||
if reader.read_line(&mut line).await.unwrap_or(0) == 0 {
|
||||
return;
|
||||
}
|
||||
if let Ok(request) = serde_json::from_str::<Value>(&line) {
|
||||
seen.lock().unwrap().push(request);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let bridge = BridgeClient::new(&addr, CONTROL_TOKEN).unwrap().with_timeout(Duration::from_millis(300));
|
||||
let error = match bridge.call_read("document.inspect", json!({})).await {
|
||||
Err(e @ AutomationError::Bridge(_)) => e.to_string(),
|
||||
other => panic!("expected the bridge timeout, got {other:?}"),
|
||||
};
|
||||
assert!(error.contains("timed out after"), "{error}");
|
||||
assert!(error.contains("not drawing frames"), "{error}");
|
||||
assert!(error.contains("a read may be retried safely"), "{error}");
|
||||
assert!(!error.contains("may have completed"), "a read timeout never claims an applied edit: {error}");
|
||||
assert_eq!(requests.lock().unwrap().len(), 1, "a timed-out call is not resent");
|
||||
}
|
||||
@@ -238,7 +238,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. 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`:
|
||||
The bridge keeps one authenticated TCP connection open, and drops and reconnects before a call once the connection has been idle long enough that the app (which closes idle connections after 30s) may have dropped it. 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. Edits are never automatically resent (#1007); a connection that could not be established is reported as "the request was not sent" instead. The MCP server opts its pure-read tools (`ui_inspect`, `doc_inspect`, `command_list`, `jobs_list`, `session_list`, `ui_screenshot`) into one automatic retry on a fresh connection, since a read applies nothing; clients driving `BridgeClient` directly opt in per call with `call_read`. The next separate call reconnects and authenticates, and reply lines whose `id` does not match the request are skipped. The client's call timeout is 55s, strictly below the server's 60s reply deadline, so the bridge's error is the one callers see; commands that can run longer should pass `"wait": false` and poll `jobs.list`. 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 \
|
||||
|
||||
Reference in new issue
Block a user