From 274f6d3e26de489107e5c8aa659e4eda2357670a Mon Sep 17 00:00:00 2001 From: Khawar Zamman Wani Date: Thu, 5 Nov 2026 18:05:13 +0530 Subject: [PATCH] MCP bridge: waited commands poll the app's job, with progress and cancel (#8130) (#8134) * MCP headless: waited commands run as jobs, with progress and cancel (#8129) Headless command_run and command_batch steps that wait (the default) ran job-capable commands inline while holding the session lock. Cancelling a long Gaussian Blur did not stop it: it still added its history step, and doc_inspect and jobs_list queued behind it. No progress was reported. A waited job-capable command now starts as a background job (Headless::command_begin) and the server polls it with the lock released (Headless::job_poll), so other calls are served meanwhile. With a progress token it sends notifications/progress (strictly increasing, at most ten a second; for a batch, finished steps plus the current step's fraction). notifications/cancelled cancels the job and leaves the document unchanged; a drop guard does the same when rmcp drops the request future. Headless keeps the outcome of each waited job as sync_jobs sees it end, since jobs.list keeps only the last few. Both backends share one batch loop, which runs no further step once the request is cancelled. Tool hints: jobs_list is read-only, doc_select is non-destructive and idempotent, and the scoped exports (which refuse to overwrite) are non-destructive. doc_save's description lost a run of spaces. Co-Authored-By: Claude Opus 5.5 * MCP headless: cancel a waited job whose request was dropped before it started The Abandon guard was created only after command_begin returned. A request dropped while the start still waited for the session lock left the job it then started running, and its `waited` entry behind. The guard is now armed inside the locked closure that starts the job, so dropping the unread output cancels it. Review feedback on #8132. Co-Authored-By: Claude Opus 5.5 * MCP headless: job_poll applies finished jobs before checking the job A caller polling Headless::job_poll directly, without sync_jobs in between, kept seeing Running after the worker finished, because the session only drops a finished job from its running list when its jobs are polled. job_poll now syncs first. The MCP path already synced in headless_op. Co-Authored-By: Claude Opus 5.5 * MCP headless: a cancel that loses the race returns the job's result job_abandon ignored whether Session::cancel_job cancelled anything and always dropped the stored outcome, so headless_wait answered `Cancelled` for a job another request had already applied to the document. It now returns how the job really ended: Cancelled only when it cancelled the job, the job's result when it had already ended, and an error when the job cannot be cancelled (printing) and is still running. Co-Authored-By: Claude Opus 5.5 * MCP headless: waited commands run as jobs, with progress and cancel (#8129) Headless command_run and command_batch steps that wait (the default) ran job-capable commands inline while holding the session lock. Cancelling a long Gaussian Blur did not stop it: it still added its history step, and doc_inspect and jobs_list queued behind it. No progress was reported. A waited job-capable command now starts as a background job (Headless::command_begin) and the server polls it with the lock released (Headless::job_poll), so other calls are served meanwhile. With a progress token it sends notifications/progress (strictly increasing, at most ten a second; for a batch, finished steps plus the current step's fraction). notifications/cancelled cancels the job and leaves the document unchanged; a drop guard does the same when rmcp drops the request future. Headless keeps the outcome of each waited job as sync_jobs sees it end, since jobs.list keeps only the last few. Both backends share one batch loop, which runs no further step once the request is cancelled. Tool hints: jobs_list is read-only, doc_select is non-destructive and idempotent, and the scoped exports (which refuse to overwrite) are non-destructive. doc_save's description lost a run of spaces. Co-Authored-By: Claude Opus 5.5 * MCP headless: cancel a waited job whose request was dropped before it started The Abandon guard was created only after command_begin returned. A request dropped while the start still waited for the session lock left the job it then started running, and its `waited` entry behind. The guard is now armed inside the locked closure that starts the job, so dropping the unread output cancels it. Review feedback on #8132. Co-Authored-By: Claude Opus 5.5 * MCP bridge: waited commands poll the app's job, with progress and cancel (#8130) A waited command_run or command_batch step of an engine command now starts it with engine.execute {wait: false} (sent once) and, when the app runs it as a background job, polls jobs.list until it ends. A job can run past the bridge's 55s reply timeout, its progress is forwarded, and cancelling or dropping the request sends jobs.cancel. UI command ids keep wait: true. A failed poll or a job no longer listed is reported with its id rather than guessed. Co-Authored-By: Claude Opus 5.5 * MCP bridge: keep the cancel guard armed until jobs.cancel returns; clarify the docs The cancel branch disarmed the Abandon guard before awaiting jobs.cancel, so a request future dropped during that await sent no cancel. The control-protocol table now says polling applies only to a waited command_run. Review feedback on #8134. Co-Authored-By: Claude Opus 5.5 * MCP headless: job_poll applies finished jobs before checking the job A caller polling Headless::job_poll directly, without sync_jobs in between, kept seeing Running after the worker finished, because the session only drops a finished job from its running list when its jobs are polled. job_poll now syncs first. The MCP path already synced in headless_op. Co-Authored-By: Claude Opus 5.5 * MCP headless: a cancel that loses the race returns the job's result job_abandon ignored whether Session::cancel_job cancelled anything and always dropped the stored outcome, so headless_wait answered `Cancelled` for a job another request had already applied to the document. It now returns how the job really ended: Cancelled only when it cancelled the job, the job's result when it had already ended, and an error when the job cannot be cancelled (printing) and is still running. Co-Authored-By: Claude Opus 5.5 * MCP bridge: a refused jobs.cancel is not reported as a cancellation bridge_wait discarded the app's reply to jobs.cancel and always answered `Cancelled`. The app refuses to cancel a job that is no longer running, so a job that finished between the jobs.list poll and the cancel was reported as cancelled although its edit may be applied. A refused cancel now answers that the job was not cancelled and points to jobs_list. Co-Authored-By: Claude Opus 5.5 * MCP bridge: a cancel during a failed poll still cancels the job When notifications/cancelled arrived while a jobs.list poll was in flight, and the poll and its retry then failed, bridge_wait disarmed the Abandon guard and returned the poll error without checking the waiter, so no jobs.cancel reached the app and the edit ran on after the caller cancelled it. That error path now sends jobs.cancel first when the request was cancelled. A failed poll without a cancel still leaves the job running and names it. Reported in review of #8134. Co-Authored-By: Claude Opus 5.5 * MCP bridge: a cancel doesn't wait for a slow jobs.list poll bridge_wait checked for a cancel only between polls, so a cancel that arrived while the app was slow to answer jobs.list waited for the poll, up to the 55s call timeout, before jobs.cancel was sent. The poll now races the cancel. That needs BridgeClient to drop the connection of a call abandoned before its reply: it kept the one cached connection, so jobs.cancel queued behind the unanswered poll on the same socket (the app serves a connection's requests in order), and a half-sent request could share it. A call now takes the connection out of the cache and returns it only once answered, as a timeout or lost reply already dropped it. The poll-error check from the previous commit stays for a cancel that races a failing poll; its test is replaced by one for the slow poll, which covers the reviewed case. A BridgeClient test checks that a dropped call's connection is not reused. Co-Authored-By: Claude Opus 5.5 * fix(automation): retain cleanup after failed bridge cancellation * chore: restart CI for PR #8134 * test(automation): classify the smart object replacement MCP tool * test(automation): synchronize dropped bridge waits and accept reconnections * Fix CI after merging upstream * Fix CI after merging upstream * Deduplicate Arabic closure messages after upstream merge * fix(engine): bound active and cancelling background workers * Fix cancellation and draining in bridge merge * Test release-aware draining cleanup * Fix draining regression warning --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Khawar Zamman Wani Co-authored-by: FredApps --- book/src/automation/mcp.md | 7 +- crates/automation/src/bridge.rs | 40 ++-- crates/automation/src/server.rs | 175 ++++++++++------ crates/automation/src/server/progress.rs | 167 +++++++++++++-- crates/automation/tests/it/bridge.rs | 36 ++++ crates/automation/tests/it/mcp.rs | 12 +- crates/automation/tests/mcp_wait.rs | 245 ++++++++++++++++++++++- docs/control-protocol.md | 4 +- docs/mcp.md | 28 +-- 9 files changed, 584 insertions(+), 130 deletions(-) diff --git a/book/src/automation/mcp.md b/book/src/automation/mcp.md index c8a942acfe..511f5f8850 100644 --- a/book/src/automation/mcp.md +++ b/book/src/automation/mcp.md @@ -63,7 +63,8 @@ Filesystem capabilities are implemented. General tool and method capabilities re The five shared core tools are `command_list`, `command_run`, `command_batch`, `doc_inspect` and `render_preview`. Existing documented tools, including `doc_render_preview`, remain -listed. `photocraft://document` and `photocraft://commands` expose live JSON. Headless -long commands that are waited on report progress and stop when the request is cancelled, and -direct headless Render Video does too, using the granted write root; see +listed. `photocraft://document` and `photocraft://commands` expose live JSON. Long commands +that are waited on report progress and stop when the request is cancelled, headless and in +bridge mode (where they may also outlast the bridge's reply timeout), and direct headless +Render Video does too, using the granted write root; see [`docs/mcp.md`](../../../docs/mcp.md) for arguments, cleanup and transport behavior. diff --git a/crates/automation/src/bridge.rs b/crates/automation/src/bridge.rs index adb6d8eea6..1f1a5e0509 100644 --- a/crates/automation/src/bridge.rs +++ b/crates/automation/src/bridge.rs @@ -1,8 +1,8 @@ //! 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, 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 +//! by `id`. Keeps one connection, reconnecting after a failure or a call +//! dropped before its reply — 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}; @@ -148,41 +148,31 @@ impl BridgeClient { *guard = Some((conn, Instant::now())); } let id = self.next_id.fetch_add(1, Ordering::Relaxed); - let Some((conn, _)) = guard.as_mut() else { + // Out of the cache until answered: a call dropped mid-exchange (a cancelled wait) closes + // its connection, so no later call shares a socket with a half-sent request or a reply + // the app is still working on. + let Some((mut conn, _)) = guard.take() else { return Err(Failure::NotSent(format!("not connected to {}", self.addr))); }; - let outcome = tokio::time::timeout(self.timeout, exchange(conn, id, method, params)).await; + let outcome = tokio::time::timeout(self.timeout, exchange(&mut 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(Failure::Definite(error)) - } + // An oversized frame leaves unread bytes. Drop this connection and + // report the budget failure without retrying a possibly completed edit. + Ok(Ok(Err(error @ AutomationError::BadRequest(_)))) => Err(Failure::Definite(error)), Ok(Ok(Ok(v))) => { - if let Some((_, last_used)) = guard.as_mut() { - *last_used = Instant::now(); - } + *guard = Some((conn, 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(); - } + *guard = Some((conn, 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)) => { - *guard = None; - Err(Failure::Transport(e.to_string())) - } - Err(_) => { - *guard = None; - Err(Failure::TimedOut(())) - } + Ok(Err(e)) => Err(Failure::Transport(e.to_string())), + Err(_) => Err(Failure::TimedOut(())), } } } diff --git a/crates/automation/src/server.rs b/crates/automation/src/server.rs index 1b125bf6be..a0ad5164ad 100644 --- a/crates/automation/src/server.rs +++ b/crates/automation/src/server.rs @@ -988,8 +988,8 @@ impl PhotocraftMcp { #[tool(description = "Run an engine command by id with JSON params (see command_list). Returns the command's JSON result. \ Long commands (filters, Content-Aware Fill/Scale, Photomerge, brush import) wait to finish unless `wait` is false: \ - then they run as a background job and the result is {job: id} (see jobs_list, jobs_cancel). Headless, a waited \ - long command reports progress and cancelling the request cancels it; other calls are served meanwhile.")] + then they run as a background job and the result is {job: id} (see jobs_list, jobs_cancel). A waited long \ + command reports progress, cancelling the request cancels it, and other calls are served meanwhile.")] async fn command_run(&self, Parameters(p): Parameters, context: RequestContext) -> Result { let mut waiter = Waiter::new(Some(context)); to_result(self.step(p.id, p.params.unwrap_or_else(|| json!({})), p.wait.unwrap_or(true), &mut waiter, (0, 1)).await) @@ -1095,7 +1095,8 @@ impl PhotocraftMcp { match &*self.backend { Backend::Headless(_) if wait => self.headless_wait(id, params, waiter, at).await.unwrap_or_else(none), Backend::Headless(_) => self.headless_op(move |h| h.command_start(&id, params, false)).await.unwrap_or_else(none), - Backend::Bridge(b) => b.call("engine.execute", json!({"command": id, "params": params, "wait": wait})).await, + Backend::Bridge(b) if wait => self.bridge_wait(b, id, params, waiter, at).await, + Backend::Bridge(b) => b.call("engine.execute", json!({"command": id, "params": params, "wait": false})).await, } } @@ -1387,65 +1388,6 @@ mod tests { assert_eq!(docs, 0, "the step after the budget ran out did not run"); } - /// Cancellation notifications only cancel the token: the handler remains alive while - /// its blocking start waits for the session lock. - #[tokio::test(flavor = "multi_thread")] - async fn cancelled_queued_commands_never_start() { - use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; - - let mcp = PhotocraftMcp::headless(); - mcp.headless_op(|h| h.command_run("file.new", json!({"width": 8, "height": 8}))).await.unwrap().unwrap(); - let before = mcp.headless_op(|h| h.inspect(None)).await.unwrap().unwrap(); - let Backend::Headless(session) = &*mcp.backend else { panic!("headless") }; - let session = session.clone(); - let (locked, acquired) = std::sync::mpsc::channel(); - let (release, released) = std::sync::mpsc::channel(); - let holder = std::thread::spawn(move || { - let _held = session.lock().unwrap(); - locked.send(()).unwrap(); - released.recv().unwrap(); - }); - acquired.recv().unwrap(); - - let (mut input, server_in) = tokio::io::duplex(65536); - let (server_out, output) = tokio::io::duplex(65536); - let server = tokio::spawn(mcp.clone().serve_io(server_in, server_out)); - let mut lines = BufReader::new(output).lines(); - async fn send(input: &mut tokio::io::DuplexStream, value: Value) { - input.write_all(format!("{value}\n").as_bytes()).await.unwrap(); - } - send(&mut input, json!({"jsonrpc":"2.0", "id":1, "method":"initialize", "params":{"protocolVersion":"2025-06-18", "capabilities":{}, "clientInfo":{"name":"test", "version":"1"}}})).await; - lines.next_line().await.unwrap().unwrap(); - send(&mut input, json!({"jsonrpc":"2.0", "method":"notifications/initialized"})).await; - for (id, command, params) in [(2, "layer.new.layer", json!({})), (3, "filter.blur.gaussianBlur", json!({"radius": 10}))] { - send( - &mut input, - json!({"jsonrpc":"2.0", "id":id, "method":"tools/call", "params":{"name":"command_run", "arguments":{"id":command, "params":params}}}), - ) - .await; - } - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - for id in [2, 3] { - send(&mut input, json!({"jsonrpc":"2.0", "method":"notifications/cancelled", "params":{"requestId":id}})).await; - } - send(&mut input, json!({"jsonrpc":"2.0", "id":4, "method":"ping"})).await; - loop { - let reply: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); - if reply["id"] == 4 { - break; - } - } - release.send(()).unwrap(); - holder.join().unwrap(); - // Let both blocked starts acquire the mutex before inspecting their effects. - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - let (after, jobs) = mcp.headless_op(|h| Ok((h.inspect(None)?, h.session.jobs_with_recent()))).await.unwrap().unwrap(); - assert_eq!(after, before); - assert!(jobs.is_empty(), "cancelled queued job started: {jobs:?}"); - drop(input); - server.await.unwrap().unwrap(); - } - /// rmcp may drop a cancelled request's future without polling it again; its job must still /// stop and leave the document unchanged (#8129). #[tokio::test(flavor = "multi_thread")] @@ -1515,6 +1457,115 @@ mod tests { assert_eq!(mcp.headless_op(|h| h.inspect(None)).await.unwrap().unwrap(), before); } + /// Cancellation notifications only cancel the token: the handler remains alive while + /// its blocking start waits for the session lock. + #[tokio::test(flavor = "multi_thread")] + async fn cancelled_queued_commands_never_start() { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + let mcp = PhotocraftMcp::headless(); + mcp.headless_op(|h| h.command_run("file.new", json!({"width": 8, "height": 8}))).await.unwrap().unwrap(); + let before = mcp.headless_op(|h| h.inspect(None)).await.unwrap().unwrap(); + let Backend::Headless(session) = &*mcp.backend else { panic!("headless") }; + let session = session.clone(); + let (locked, acquired) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + let holder = std::thread::spawn(move || { + let _held = session.lock().unwrap(); + locked.send(()).unwrap(); + released.recv().unwrap(); + }); + acquired.recv().unwrap(); + + let (mut input, server_in) = tokio::io::duplex(65536); + let (server_out, output) = tokio::io::duplex(65536); + let server = tokio::spawn(mcp.clone().serve_io(server_in, server_out)); + let mut lines = BufReader::new(output).lines(); + async fn send(input: &mut tokio::io::DuplexStream, value: Value) { + input.write_all(format!("{value}\n").as_bytes()).await.unwrap(); + } + send(&mut input, json!({"jsonrpc":"2.0", "id":1, "method":"initialize", "params":{"protocolVersion":"2025-06-18", "capabilities":{}, "clientInfo":{"name":"test", "version":"1"}}})).await; + lines.next_line().await.unwrap().unwrap(); + send(&mut input, json!({"jsonrpc":"2.0", "method":"notifications/initialized"})).await; + for (id, command, params) in [(2, "layer.new.layer", json!({})), (3, "filter.blur.gaussianBlur", json!({"radius": 10}))] { + send( + &mut input, + json!({"jsonrpc":"2.0", "id":id, "method":"tools/call", "params":{"name":"command_run", "arguments":{"id":command, "params":params}}}), + ) + .await; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + for id in [2, 3] { + send(&mut input, json!({"jsonrpc":"2.0", "method":"notifications/cancelled", "params":{"requestId":id}})).await; + } + send(&mut input, json!({"jsonrpc":"2.0", "id":4, "method":"ping"})).await; + loop { + let reply: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + if reply["id"] == 4 { + break; + } + } + release.send(()).unwrap(); + holder.join().unwrap(); + // Let both blocked starts acquire the mutex before inspecting their effects. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let (after, jobs) = mcp.headless_op(|h| Ok((h.inspect(None)?, h.session.jobs_with_recent()))).await.unwrap().unwrap(); + assert_eq!(after, before); + assert!(jobs.is_empty(), "cancelled queued job started: {jobs:?}"); + drop(input); + server.await.unwrap().unwrap(); + } + + /// The same over the bridge (#8130): a dropped wait sends `jobs.cancel` for the app's job. + #[tokio::test(flavor = "multi_thread")] + async fn dropping_a_bridged_wait_cancels_the_apps_job() { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; + const TOKEN: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let (cancels, mut cancelled) = tokio::sync::mpsc::unbounded_channel(); + let polled = Arc::new(tokio::sync::Notify::new()); + let polling = polled.clone(); + // A fake app whose job 3 never ends and whose poll never replies. Dropping the wait + // discards that connection, so the cancellation must arrive on a fresh one (#8130). + let app = tokio::spawn(async move { + loop { + let (socket, _) = listener.accept().await.unwrap(); + let (read, mut write) = socket.into_split(); + let mut lines = tokio::io::BufReader::new(read).lines(); + let (auth, _) = crate::security::authentication_reply(&lines.next_line().await.unwrap().unwrap(), TOKEN); + write.write_all(format!("{auth}\n").as_bytes()).await.unwrap(); + while let Ok(Some(line)) = lines.next_line().await { + let request: Value = serde_json::from_str(&line).unwrap(); + let result = match request["method"].as_str().unwrap() { + "engine.execute" => json!({"job": 3, "pending": true}), + "jobs.list" => { + polling.notify_one(); + continue; + } + "jobs.cancel" => json!({"cancelled": 1}), + method => panic!("unexpected method {method}"), + }; + write.write_all(format!("{}\n", json!({"id": request["id"], "ok": true, "result": result})).as_bytes()).await.unwrap(); + if request["method"] == "jobs.cancel" { + cancels.send(request).unwrap(); + return; + } + } + } + }); + let mcp = PhotocraftMcp::with_backend(Backend::Bridge(Arc::new(BridgeClient::new(addr, TOKEN).unwrap()))); + let mut waited = Box::pin(mcp.run_command("filter.blur.gaussianBlur".into(), json!({"radius": 4}), true)); + tokio::select! { + result = &mut waited => panic!("the running job answered before its wait was dropped: {result:?}"), + ready = tokio::time::timeout(std::time::Duration::from_secs(10), polled.notified()) => ready.expect("the app received the poll"), + } + drop(waited); + let cancel = tokio::time::timeout(std::time::Duration::from_secs(10), cancelled.recv()).await.unwrap().unwrap(); + assert_eq!((&cancel["method"], &cancel["params"]), (&json!("jobs.cancel"), &json!({"job": 3}))); + tokio::time::timeout(std::time::Duration::from_secs(10), app).await.unwrap().unwrap(); + } + #[tokio::test(flavor = "multi_thread")] async fn mcp_doc_open_schema_and_structured_error() { let schema = &PhotocraftMcp::tool_router().map["doc_open"].attr.input_schema; diff --git a/crates/automation/src/server/progress.rs b/crates/automation/src/server/progress.rs index 270694d523..8b3697f65d 100644 --- a/crates/automation/src/server/progress.rs +++ b/crates/automation/src/server/progress.rs @@ -1,6 +1,7 @@ //! Progress and cancellation for calls that wait. Render Video runs on the existing blocking -//! worker; other long commands run as engine jobs that the session polls ([`Waiter`]). -use super::{Arc, AutomationError, Backend, CallToolResult, Headless, Mutex, PhotocraftMcp, PoisonError, Value, fail, no_backend, ok_json}; +//! worker; other long commands run as engine jobs that the server polls ([`Waiter`]): the headless +//! session's own, or the app's over the bridge. +use super::{Arc, AutomationError, Backend, BridgeClient, CallToolResult, PhotocraftMcp, PoisonError, Value, fail, json, no_backend, ok_json}; use crate::headless::JobPoll; use photocraft_engine::jobs::{JobId, Started}; use rmcp::RoleServer; @@ -91,19 +92,31 @@ impl Waiter { /// Cancels a waited job when its waiter goes away without collecting it: rmcp may drop the /// request future on cancellation, and nothing else would stop the job. struct Abandon { - session: Arc>, + backend: Arc, job: Option, } impl Drop for Abandon { fn drop(&mut self) { let Some(job) = self.job.take() else { return }; - let session = self.session.clone(); - let abandon = move || drop(session.lock().unwrap_or_else(PoisonError::into_inner).job_abandon(job)); - match tokio::runtime::Handle::try_current() { - // A synchronous command can hold the lock for a while: don't block a runtime thread. - Ok(runtime) => drop(runtime.spawn_blocking(abandon)), - Err(_) => abandon(), + let runtime = tokio::runtime::Handle::try_current(); + match &*self.backend { + Backend::Headless(session) => { + let session = session.clone(); + let abandon = move || drop(session.lock().unwrap_or_else(PoisonError::into_inner).job_abandon(job)); + match runtime { + // A synchronous command can hold the lock for a while: don't block a runtime thread. + Ok(runtime) => drop(runtime.spawn_blocking(abandon)), + Err(_) => abandon(), + } + } + // Without a runtime there is no way to reach the app; the job then runs to the end. + Backend::Bridge(client) => { + if let Ok(runtime) = runtime { + let client = client.clone(); + drop(runtime.spawn(async move { client.call("jobs.cancel", json!({"job": job.0})).await })); + } + } } } } @@ -114,8 +127,10 @@ impl PhotocraftMcp { /// `waiter` as `step.0 + fraction` of `step.1`, and cancelling the request cancels the job, /// leaving the document unchanged. `None` in bridge mode. pub(super) async fn headless_wait(&self, id: String, params: Value, waiter: &mut Waiter, step: (usize, usize)) -> Option> { - let Backend::Headless(session) = &*self.backend else { return None }; - let session = session.clone(); + if !matches!(&*self.backend, Backend::Headless(_)) { + return None; + } + let backend = self.backend.clone(); let cancel = waiter.context.as_ref().map(|context| context.ct.clone()); // The guard is armed under the lock that starts the job: when the request is dropped while // the start still waits for the lock, the closure's output is dropped unread and cancels it. @@ -128,7 +143,7 @@ impl PhotocraftMcp { Started::Job(job) if cancel.as_ref().is_some_and(|token| token.is_cancelled()) => { return h.job_abandon(job).map(Err); } - Started::Job(job) => Ok(Abandon { session, job: Some(job) }), + Started::Job(job) => Ok(Abandon { backend, job: Some(job) }), Started::Done(value) => Err(value), }) }) @@ -165,4 +180,132 @@ impl PhotocraftMcp { delay = (delay * 2).min(Duration::from_millis(50)); } } + /// Bridge: run command `id` for a caller that waits (#8130). An engine command starts with + /// `wait: false`; when the app runs it as a background job, `jobs.list` is polled until the job + /// ends, so no single reply has to arrive within the client's call timeout. Progress and + /// cancellation as in [`Self::headless_wait`]. Only the start is an edit, and it is sent once + /// (#1007); a failed poll reports the job instead of guessing its outcome. + pub(super) async fn bridge_wait( + &self, + client: &BridgeClient, + id: String, + params: Value, + waiter: &mut Waiter, + step: (usize, usize), + ) -> Result { + // A UI command can start a job without naming it in its reply: the app waits on those. + if photocraft_engine::commands::find(&id).is_none() { + return client.call("engine.execute", json!({"command": id, "params": params, "wait": true})).await; + } + let started = client.call("engine.execute", json!({"command": id, "params": params, "wait": false})).await?; + let pending = started.get("pending").and_then(Value::as_bool) == Some(true); + let Some(job) = started.get("job").and_then(Value::as_u64).filter(|_| pending) else { + return Ok(started); + }; + let mut guard = Abandon { backend: self.backend.clone(), job: Some(JobId(job)) }; + let mut delay = Duration::from_millis(10); + loop { + // Race the poll against a cancel: an app slow to answer must not hold it until the call times out. + let polled = tokio::select! { + biased; + listed = client.call_read("jobs.list", json!({})) => Some(listed), + () = waiter.cancelled() => None, + }; + let listed = match polled { + Some(Ok(listed)) => listed, + Some(Err(error)) if !waiter.is_cancelled() => { + // The app may be fine and the job still running: leave it, and say how to find it. + guard.job = None; + return Err(AutomationError::Bridge(format!( + "lost track of job {job} ({id}): {error}; it may still be running and apply its result: check jobs_list, or stop it with jobs_cancel" + ))); + } + // Cancelled during the poll, which may also have failed: the app may answer a + // fresh connection. + _ => return Err(bridge_cancel(client, &mut guard, job, &id).await), + }; + let entry = listed.get("jobs").and_then(Value::as_array).and_then(|jobs| jobs.iter().find(|j| j.get("id").and_then(Value::as_u64) == Some(job))); + let Some(entry) = entry else { + guard.job = None; + return Err(AutomationError::Bridge(format!( + "job {job} ({id}) ended but jobs.list no longer shows it (it keeps the last few); inspect the document for its result" + ))); + }; + let text = |key: &str| entry.get(key).and_then(Value::as_str).unwrap_or_default().to_owned(); + match entry.get("state").and_then(Value::as_str) { + Some("running") => { + let fraction = entry.get("progress").and_then(Value::as_f64).filter(|f| f.is_finite()).map_or(0.0, |f| f.clamp(0.0, 1.0)); + waiter.report(step.0 as f64 + fraction, step.1 as f64, &text("message")).await; + } + state => { + guard.job = None; + // The same results the app's own waited reply gives. + return match state { + Some("done") => Ok(entry.get("result").cloned().unwrap_or(Value::Null)), + Some("failed") => Err(AutomationError::App(Some(text("error")).filter(|e| !e.is_empty()).unwrap_or_else(|| "unknown error".into()))), + Some("cancelled") => Err(AutomationError::App("cancelled".into())), + _ => Err(AutomationError::Bridge(format!("job {job} ({id}) has an unknown state: {entry}"))), + }; + } + } + tokio::select! { + biased; + () = waiter.cancelled() => return Err(bridge_cancel(client, &mut guard, job, &id).await), + () = tokio::time::sleep(delay) => {} + } + delay = (delay * 2).min(Duration::from_millis(200)); + } + } +} + +/// Cancel bridged `job` (`id`) before answering a cancelled wait. Failed calls leave the guard +/// armed for one best-effort cleanup attempt on drop. +async fn bridge_cancel(client: &BridgeClient, guard: &mut Abandon, job: u64, id: &str) -> AutomationError { + let cancel = client.call("jobs.cancel", json!({"job": job})).await; + // A failed or abandoned call keeps the guard armed for one cleanup attempt on drop. + // Only an acknowledged cancellation proves that cleanup is no longer needed. + // A definite refusal is not improved by retrying; malformed replies and transport + // failures leave the guard armed because the outcome is uncertain. + if cancel.as_ref().is_ok_and(|value| cancellation_acknowledged(value) || refusal(value, job).is_some()) { + guard.job = None; + } + cancel_answer(job, id, cancel) +} + +/// The answer to a bridged wait that was cancelled, from the app's reply to `jobs.cancel`. The app +/// refuses to cancel a job that already ended, whose result may then be applied: say so rather +/// than report a cancellation. +fn cancel_answer(job: u64, id: &str, reply: Result) -> AutomationError { + match reply { + Ok(value) if refusal(&value, job).is_some() => { + let reason = refusal(&value, job).unwrap_or("unknown reason"); + AutomationError::Bridge(format!("job {job} ({id}) was not cancelled: {reason}; check jobs_list for its outcome")) + } + Ok(value) if cancellation_acknowledged(&value) => photocraft_engine::EngineError::Cancelled.into(), + Ok(value) => AutomationError::Bridge(format!("cancellation of job {job} ({id}) was not confirmed: {value}; check jobs_list for its outcome")), + Err(error) => AutomationError::Bridge(format!("job {job} ({id}) cancellation was not confirmed: {error}; check jobs_list for its outcome")), + } +} + +fn cancellation_acknowledged(value: &Value) -> bool { + value.get("cancelled").and_then(Value::as_u64) == Some(1) +} + +fn refusal(value: &Value, job: u64) -> Option<&str> { + value.get("notCancelled")?.as_array()?.iter().find(|entry| entry.get("job").and_then(Value::as_u64) == Some(job))?.get("reason")?.as_str() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_refused_cancel_is_not_reported_as_cancelled() { + let cancelled = cancel_answer(7, "filter.blur.gaussianBlur", Ok(json!({"cancelled": 1}))); + assert!(matches!(cancelled, AutomationError::Engine(photocraft_engine::EngineError::Cancelled)), "{cancelled}"); + let refused = cancel_answer(7, "filter.blur.gaussianBlur", Err(AutomationError::App("job 7 is not running".into()))); + assert!(matches!(&refused, AutomationError::Bridge(m) if m.contains("not confirmed") && m.contains("job 7 is not running")), "{refused}"); + let refused = cancel_answer(7, "filter.blur.gaussianBlur", Ok(json!({"cancelled": 0, "notCancelled": [{"job": 7, "reason": "committing"}]}))); + assert!(matches!(&refused, AutomationError::Bridge(m) if m.contains("job 7") && m.contains("committing") && m.contains("jobs_list")), "{refused}"); + } } diff --git a/crates/automation/tests/it/bridge.rs b/crates/automation/tests/it/bridge.rs index 258e08e4fd..460a33e264 100644 --- a/crates/automation/tests/it/bridge.rs +++ b/crates/automation/tests/it/bridge.rs @@ -164,3 +164,39 @@ async fn a_timed_out_call_is_not_retried() { 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"); } + +#[tokio::test(flavor = "multi_thread")] +async fn a_call_dropped_before_its_reply_does_not_reuse_its_connection() { + // Connection 1 never answers. A call dropped while waiting on it (a cancelled MCP wait) must + // close it: on the same socket the next request would queue behind the unanswered one. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let accepted = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count = std::sync::Arc::clone(&accepted); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let silent = count.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0; + tokio::spawn(async move { + authenticate(&mut socket).await; + let (read, mut write) = socket.into_split(); + let mut lines = BufReader::new(read).lines(); + while let Ok(Some(line)) = lines.next_line().await { + let Ok(request) = serde_json::from_str::(&line) else { continue }; + if !silent { + let reply = json!({"id": request["id"], "ok": true, "result": {"served": true}}); + if write.write_all(format!("{reply}\n").as_bytes()).await.is_err() { + return; + } + } + } + }); + } + }); + + let bridge = BridgeClient::new(&addr, CONTROL_TOKEN).unwrap().with_timeout(Duration::from_secs(30)); + let dropped = tokio::time::timeout(Duration::from_millis(300), bridge.call_read("jobs.list", json!({}))).await; + assert!(dropped.is_err(), "the silent connection answered: {dropped:?}"); + let next = tokio::time::timeout(Duration::from_secs(10), bridge.call("jobs.cancel", json!({"job": 7}))).await; + assert_eq!(next.expect("the next call waited on the dropped call's connection").unwrap()["served"], true); + assert_eq!(accepted.load(std::sync::atomic::Ordering::SeqCst), 2, "the next call opened a fresh connection"); +} diff --git a/crates/automation/tests/it/mcp.rs b/crates/automation/tests/it/mcp.rs index b57b77db4d..b2746303b8 100644 --- a/crates/automation/tests/it/mcp.rs +++ b/crates/automation/tests/it/mcp.rs @@ -1305,10 +1305,14 @@ async fn command_batch_step_without_waiting_starts_a_background_job() { async fn bridge_command_batch_forwards_each_steps_wait() { let (addr, app) = fake_app().await; let client = connect(PhotocraftMcp::bridge(&addr, CONTROL_TOKEN).unwrap()).await; - let r = json_of(&call(&client, "command_batch", json!({"steps": [{"id": "filter.blur.gaussianBlur", "wait": false}, {"id": "layer.new.layer"}]})).await); - assert_eq!(r["completed"], 2, "{r}"); + let steps = json!([{"id": "filter.blur.gaussianBlur", "wait": false}, {"id": "layer.new.layer"}, {"id": "view.zoomIn"}]); + let r = json_of(&call(&client, "command_batch", json!({"steps": steps})).await); + assert_eq!(r["completed"], 3, "{r}"); assert_eq!(r["results"][0]["result"]["wait"], false, "{r}"); - assert_eq!(r["results"][1]["result"]["wait"], true, "{r}"); + // A waited engine command starts without waiting too: the bridge polls a job it starts (#8130). + assert_eq!(r["results"][1]["result"]["wait"], false, "{r}"); + // A UI command can't be polled, so the app waits on it. + assert_eq!(r["results"][2]["result"]["wait"], true, "{r}"); client.cancel().await.unwrap(); app.abort(); } @@ -1414,10 +1418,10 @@ async fn convention_tool_hints_match_what_each_tool_does() { ("doc_export_data_sets", ADDS), ("doc_select", (false, false, true)), ("doc_save", EDITS), + ("doc_replace_smart_object", EDITS), ("doc_export", EDITS), ("doc_close", EDITS), ("doc_import_data_sets", EDITS), - ("doc_replace_smart_object", EDITS), ("command_run", EDITS), ("command_batch", EDITS), ("jobs_cancel", EDITS), diff --git a/crates/automation/tests/mcp_wait.rs b/crates/automation/tests/mcp_wait.rs index c78bbb3680..797b1d141a 100644 --- a/crates/automation/tests/mcp_wait.rs +++ b/crates/automation/tests/mcp_wait.rs @@ -1,8 +1,11 @@ -//! Headless calls that wait on a long command (#8129): progress, other calls served meanwhile, -//! and cancellation that leaves the document unchanged. Driven over the real line transport. -use std::time::Duration; +//! Calls that wait on a long command: headless (#8129), with progress, other calls served +//! meanwhile and cancellation that leaves the document unchanged; and over the bridge (#8130), +//! polling the app's job past the reply timeout. Driven over the real line transport. +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; -use photocraft_automation::PhotocraftMcp; +use photocraft_automation::security::authentication_reply; +use photocraft_automation::{Backend, BridgeClient, PhotocraftMcp}; use serde_json::{Value, json}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, Lines}; @@ -13,10 +16,10 @@ struct Wire { } impl Wire { - async fn start() -> Wire { + async fn start(mcp: PhotocraftMcp) -> Wire { let (input, server_in) = tokio::io::duplex(1 << 20); let (server_out, output) = tokio::io::duplex(1 << 20); - let server = tokio::spawn(PhotocraftMcp::headless().serve_io(server_in, server_out)); + let server = tokio::spawn(mcp.serve_io(server_in, server_out)); let mut wire = Wire { input, lines: BufReader::new(output).lines(), server }; wire.send(json!({"jsonrpc":"2.0", "id":1, "method":"initialize", "params":{"protocolVersion":"2025-06-18", "capabilities":{}, "clientInfo":{"name":"test", "version":"1"}}})) .await; @@ -44,14 +47,21 @@ impl Wire { /// The reply to `id`, skipping progress notes. `pending` must not answer meanwhile. async fn reply(&mut self, id: u64, pending: Option) -> Value { + let (text, error) = self.outcome(id, pending).await; + assert!(!error, "{text}"); + serde_json::from_str(&text).unwrap() + } + + /// The text of the tool result for `id` and whether it is an error. + async fn outcome(&mut self, id: u64, pending: Option) -> (String, bool) { loop { let message = self.next().await; if pending.is_some_and(|p| message["id"] == p) { panic!("request {} answered while it should still run: {message}", pending.unwrap_or(0)); } if message["id"] == id { - assert_eq!(message["result"]["isError"], false, "{message}"); - return serde_json::from_str(message["result"]["content"][0]["text"].as_str().unwrap()).unwrap(); + let text = message["result"]["content"][0]["text"].as_str().unwrap_or_else(|| panic!("{message}")).to_owned(); + return (text, message["result"]["isError"] == true); } } } @@ -109,7 +119,7 @@ const BLUR: fn() -> Value = || json!({"id": "filter.blur.gaussianBlur", "params" #[tokio::test(flavor = "multi_thread")] async fn a_waited_command_reports_progress_serves_other_calls_and_cancels_cleanly() { - let mut wire = Wire::start().await; + let mut wire = Wire::start(PhotocraftMcp::headless()).await; wire.big_document().await; let before = wire.history(4).await; @@ -139,7 +149,7 @@ async fn a_waited_command_reports_progress_serves_other_calls_and_cancels_cleanl #[tokio::test(flavor = "multi_thread")] async fn cancelling_a_batch_cancels_its_waited_step_and_runs_no_later_step() { - let mut wire = Wire::start().await; + let mut wire = Wire::start(PhotocraftMcp::headless()).await; wire.big_document().await; let before = wire.history(4).await; @@ -154,6 +164,221 @@ async fn cancelling_a_batch_cancels_its_waited_step_and_runs_no_later_step() { wire.finish().await; } +const TOKEN: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +/// A scripted desktop app for bridge mode. A blur runs as job 7 for `runs`: `engine.execute` +/// with `wait: false` answers `{job, pending}` at once, without it only when the job ends, like +/// the app. `jobs.list` reports the job and `jobs.cancel` stops it. Other commands answer at +/// once. Every request is logged. +struct FakeApp { + runs: Duration, + /// Close the connection instead of answering the start (its reply is lost). + lose_start: bool, + /// Lose the first cancellation request before applying it; a later request can succeed. + lose_cancel: bool, + /// How long to hold each `jobs.list` reply (an app slow to answer). + hold_polls: Duration, + started: Option, + cancelled: bool, + log: Vec, +} + +impl FakeApp { + async fn serve(runs: Duration, lose_start: bool) -> (String, Arc>) { + let app = Arc::new(Mutex::new(FakeApp { + runs, + lose_start, + lose_cancel: false, + hold_polls: Duration::ZERO, + started: None, + cancelled: false, + log: Vec::new(), + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let shared = app.clone(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let app = shared.clone(); + tokio::spawn(async move { + let (read, mut write) = socket.into_split(); + let mut lines = BufReader::new(read).lines(); + let (auth, ok) = authentication_reply(&lines.next_line().await.unwrap().unwrap(), TOKEN); + assert!(ok, "each connection must authenticate"); + write.write_all(format!("{auth}\n").as_bytes()).await.unwrap(); + while let Ok(Some(line)) = lines.next_line().await { + let request: Value = serde_json::from_str(&line).unwrap(); + let Some((result, after)) = app.lock().unwrap().answer(&request) else { return }; + tokio::time::sleep(after).await; + let reply = json!({"id": request["id"], "ok": true, "result": result}); + if write.write_all(format!("{reply}\n").as_bytes()).await.is_err() { + return; + } + } + }); + } + }); + (addr, app) + } + + /// The result for `request` and how long to hold it; `None` closes the connection. + fn answer(&mut self, request: &Value) -> Option<(Value, Duration)> { + self.log.push(request.clone()); + let p = &request["params"]; + let result = match request["method"].as_str().unwrap() { + "engine.execute" if p["wait"] == false && p["command"] == "filter.blur.gaussianBlur" => { + if self.lose_start { + return None; + } + self.started = Some(Instant::now()); + json!({"job": 7, "pending": true}) + } + "engine.execute" if p["command"] == "filter.blur.gaussianBlur" => { + self.started = Some(Instant::now()); + return Some((json!({"blurred": true}), self.runs)); + } + "engine.execute" => json!({"ran": p["command"]}), + "jobs.list" => { + let Some(started) = self.started else { return Some((json!({"jobs": []}), Duration::ZERO)) }; + let progress = started.elapsed().as_secs_f64() / self.runs.as_secs_f64(); + let job = match () { + () if self.cancelled => json!({"id": 7, "state": "cancelled", "progress": progress.min(1.0), "message": ""}), + () if progress >= 1.0 => json!({"id": 7, "state": "done", "progress": 1.0, "message": "", "result": {"blurred": true}}), + () => json!({"id": 7, "state": "running", "progress": progress, "message": "Blurring"}), + }; + return Some((json!({"jobs": [job]}), self.hold_polls)); + } + "jobs.cancel" => { + if self.lose_cancel { + self.lose_cancel = false; + return None; + } + self.cancelled = true; + json!({"cancelled": 1}) + } + method => panic!("unexpected `{method}`"), + }; + Some((result, Duration::ZERO)) + } + + /// The logged requests for `method`. + fn sent(app: &Mutex, method: &str) -> Vec { + app.lock().unwrap().log.iter().filter(|r| r["method"] == method).map(|r| r["params"].clone()).collect() + } +} + +/// An MCP server bridged to `addr` whose every call times out after 300ms. +fn bridged(addr: &str) -> PhotocraftMcp { + PhotocraftMcp::with_backend(Backend::Bridge(Arc::new(BridgeClient::new(addr, TOKEN).unwrap().with_timeout(Duration::from_millis(300))))) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_bridged_job_that_outlives_the_reply_timeout_returns_its_result_with_progress() { + // The job runs five times as long as any one reply may take. + let (addr, app) = FakeApp::serve(Duration::from_millis(1500), false).await; + let mut wire = Wire::start(bridged(&addr)).await; + wire.call(5, "command_run", json!({"id": "filter.blur.gaussianBlur", "params": {"radius": 9}}), Some("blur")).await; + let progress = wire.first_progress(5, "blur", 1.0).await; + assert!((0.0..1.0).contains(&progress), "{progress}"); + assert_eq!(wire.reply(5, None).await, json!({"blurred": true})); + // Before #8130 the edit was sent waiting, and the call failed after the timeout. + assert_eq!(FakeApp::sent(&app, "engine.execute"), [json!({"command": "filter.blur.gaussianBlur", "params": {"radius": 9}, "wait": false})]); + assert!(FakeApp::sent(&app, "jobs.list").len() > 1); + assert!(FakeApp::sent(&app, "jobs.cancel").is_empty()); + + // A command the engine doesn't know (a UI command) still lets the app wait. + assert!(photocraft_engine::commands::find("view.zoomIn").is_none()); + wire.call(6, "command_run", json!({"id": "view.zoomIn"}), None).await; + wire.reply(6, None).await; + assert_eq!(FakeApp::sent(&app, "engine.execute").last(), Some(&json!({"command": "view.zoomIn", "params": {}, "wait": true}))); + + // So does every waited step of a batch, which reports the steps around the job's progress. + let steps = json!([{"id": "filter.blur.gaussianBlur", "params": {"radius": 9}}, {"id": "layer.new.layer"}]); + app.lock().unwrap().started = None; + wire.call(7, "command_batch", json!({"steps": steps}), Some("batch")).await; + assert!(wire.first_progress(7, "batch", 2.0).await < 1.0, "still in the first step"); + let batch = wire.reply(7, None).await; + assert_eq!(batch["completed"], 2, "{batch}"); + assert_eq!(batch["results"][0]["result"], json!({"blurred": true}), "{batch}"); + wire.finish().await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn cancelling_a_bridged_wait_cancels_the_apps_job() { + let (addr, app) = FakeApp::serve(Duration::from_secs(600), false).await; + let mut wire = Wire::start(bridged(&addr)).await; + wire.call(5, "command_run", BLUR(), Some("blur")).await; + wire.first_progress(5, "blur", 1.0).await; + wire.cancel(5).await; + let t = Instant::now(); + while FakeApp::sent(&app, "jobs.cancel").is_empty() { + assert!(t.elapsed() < Duration::from_secs(10), "the job was never cancelled"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(FakeApp::sent(&app, "jobs.cancel"), [json!({"job": 7})]); + assert_eq!(FakeApp::sent(&app, "engine.execute").len(), 1, "the edit is sent once"); + wire.finish().await; +} + +/// A failed cancellation keeps the drop guard armed, so its cleanup can reach the app. +#[tokio::test(flavor = "multi_thread")] +async fn a_failed_bridge_cancel_keeps_cleanup_armed() { + let (addr, app) = FakeApp::serve(Duration::from_secs(600), false).await; + app.lock().unwrap().lose_cancel = true; + let mut wire = Wire::start(bridged(&addr)).await; + wire.call(5, "command_run", BLUR(), Some("blur")).await; + wire.first_progress(5, "blur", 1.0).await; + wire.cancel(5).await; + + let deadline = Instant::now() + Duration::from_secs(10); + while !app.lock().unwrap().cancelled { + assert!(Instant::now() < deadline, "the failed jobs.cancel disabled cleanup and left the job running"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(FakeApp::sent(&app, "jobs.cancel"), [json!({"job": 7}), json!({"job": 7})]); + assert_eq!(FakeApp::sent(&app, "engine.execute").len(), 1, "the edit is never resent"); + wire.finish().await; +} + +/// A cancel during a poll the app is slow to answer, or that then fails (review of #8134), reaches +/// the app at once over a fresh connection. +#[tokio::test(flavor = "multi_thread")] +async fn cancelling_during_a_slow_poll_cancels_without_waiting_for_it() { + let (addr, app) = FakeApp::serve(Duration::from_secs(600), false).await; + app.lock().unwrap().hold_polls = Duration::from_secs(600); + // The poll would hold the cancel for this long if it waited for it. + let bridge = BridgeClient::new(&addr, TOKEN).unwrap().with_timeout(Duration::from_secs(30)); + let mut wire = Wire::start(PhotocraftMcp::with_backend(Backend::Bridge(Arc::new(bridge)))).await; + wire.call(5, "command_run", BLUR(), Some("blur")).await; + let t = Instant::now(); + while FakeApp::sent(&app, "jobs.list").is_empty() { + assert!(t.elapsed() < Duration::from_secs(10), "the job was never polled"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + wire.cancel(5).await; + let t = Instant::now(); + while FakeApp::sent(&app, "jobs.cancel").is_empty() { + assert!(t.elapsed() < Duration::from_secs(10), "the cancel waited for the poll"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(FakeApp::sent(&app, "jobs.cancel"), [json!({"job": 7})]); + assert_eq!(FakeApp::sent(&app, "jobs.list").len(), 1, "the slow poll was not retried"); + wire.finish().await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_lost_start_reply_is_reported_and_never_resent() { + let (addr, app) = FakeApp::serve(Duration::from_millis(100), true).await; + let mut wire = Wire::start(bridged(&addr)).await; + wire.call(5, "command_run", BLUR(), None).await; + let (error, is_error) = wire.outcome(5, None).await; + assert!(is_error && error.contains("may have completed"), "{error}"); + // No resend of the edit, and no polling for a job whose id never arrived. + assert_eq!(FakeApp::sent(&app, "engine.execute").len(), 1); + assert!(FakeApp::sent(&app, "jobs.list").is_empty()); + wire.finish().await; +} + #[test] fn polling_a_job_directly_returns_its_result_once_it_ends() { use photocraft_automation::headless::JobPoll; diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 5f69c63122..9ea0d5b08e 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -361,7 +361,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. CPU composition is the default; `--gpu` or `PHOTOCRAFT_GPU=1` enables optional GPU rendering with automatic CPU fallback. - **Bridge** (`photocraft-cli mcp --bridge 127.0.0.1:7878 --control-token-file `): every tool is forwarded to a running `photocraft --control 7878 --control-token-file ` over this protocol, so agents see and drive the live app. Explicit `--gpu` / `--no-gpu` flags are rejected with a usage error because the desktop controls rendering; `PHOTOCRAFT_GPU` is ignored in bridge mode. `--bridge 7878` is short for `--bridge 127.0.0.1:7878`. With no `--bridge`, `PHOTOCRAFT_CONTROL_PORT` selects the bridge port (`127.0.0.1:`), and the CLI says so on standard error; with neither (or the variable empty), the MCP runs headless. A value that is not a port is a usage error naming the variable. `--automation-read-root` or `--automation-write-root` with the variable set is a usage error, since the bridge would drop the roots. -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`: +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 (a call abandoned before its reply, such as a cancelled MCP wait, also drops it, so the next call never queues behind it); 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. A waited MCP `command_run` (or `command_batch` step) of an engine command is not bound by it (#8130): the bridge sends `engine.execute` once with `"wait": false` and, when the app starts a background job, polls `jobs.list` until the job ends and returns its result, forwarding progress and cancelling the job (`jobs.cancel`) when the request is cancelled, at once, even while a poll still waits on the app. Limits: a command that does not run as a job, a UI command (not an engine command; the app waits on it) and a lost reply to the start itself still fail at the timeout, the last with "operation may have completed" (look for the job in `jobs_list` before retrying); a failed poll of an uncancelled wait reports the job id and leaves the job running; and an unresponsive app still times out. Clients driving `BridgeClient` directly pass `"wait": false` and poll `jobs.list` themselves. 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 \ @@ -372,7 +372,7 @@ How each MCP tool maps onto control methods in bridge mode: | MCP tool | Control method | |---|---| -| `command_run {id, params, wait?}` | `engine.execute {command: id, params, wait}` | +| `command_run {id, params, wait?}` | `engine.execute {command: id, params, wait: false}`; when `wait` is true (the default), then `jobs.list` until a started job ends (`jobs.cancel` on cancellation), and with `wait: false` the `{job, pending}` reply is returned as is. A UI command id is sent with `wait` as given | | `jobs_list` / `jobs_cancel {job?}` | `jobs.list` / `jobs.cancel {job?}` | | `command_list {filter?, enabled_only?}` | `engine.commands` (filtered by the MCP server; replies with the same bare array, rows keep `disabledReason` and `needsDocument`, see [Command list replies](#command-list-replies)) | | `doc_new {…}` | `engine.execute {command: "file.new", params}` | diff --git a/docs/mcp.md b/docs/mcp.md index 60c6320741..0177a9f4ae 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -137,18 +137,22 @@ for ten minutes. List/read responses include the MCP 2026-07-28 result/cache fie ## Progress and cancellation -Headless `command_run` (and each `command_batch` step) that waits, the default, runs a -job-capable command (filters, Content-Aware Fill/Scale, Photomerge, …) as a background job -and returns its result when the job ends. Meanwhile the session serves other calls -(`doc_inspect`, `jobs_list`, …), and the running job shows in `jobs_list`. With -`_meta.progressToken`, the call sends `notifications/progress` with the job's progress (0–1, -`total` 1; for a batch, finished steps plus the current step's fraction, `total` the step -count) and the job's `message`, increasing strictly and at most ten per second. -`notifications/cancelled`, or the client dropping the request, cancels the job: the document -stays unchanged and a cancelled batch runs no further step. A job that already finished -before the cancellation arrived stays applied. Commands that are not jobs still run to the -end. Bridge mode still waits on the app's reply (#8130); a bridge `command_batch` reports -finished steps as progress and sends no further step once the request is cancelled. +`command_run` (and each `command_batch` step) that waits, the default, runs a job-capable +command (filters, Content-Aware Fill/Scale, Photomerge, …) as a background job and returns its +result when the job ends. Meanwhile the server serves other calls (`doc_inspect`, `jobs_list`, +…), and the running job shows in `jobs_list`. With `_meta.progressToken`, the call sends +`notifications/progress` with the job's progress (0–1, `total` 1; for a batch, finished steps +plus the current step's fraction, `total` the step count) and the job's `message`, increasing +strictly and at most ten per second. `notifications/cancelled`, or the client dropping the +request, cancels the job: the document stays unchanged and a cancelled batch runs no further +step. A job that already finished before the cancellation arrived stays applied. Commands that +are not jobs still run to the end. + +Headless, the session runs the job itself (#8129). In bridge mode (#8130) the app runs it: the +server starts it with `engine.execute {wait: false}` (sent once, never resent) and polls +`jobs.list`, so a job may run past the bridge's 55s reply timeout. A command that is not a job, +a UI command id and a lost reply to the start still fail at that timeout; see +[MCP bridge](control-protocol.md#mcp-bridge) for the limits. A direct headless `command_run` of `file.export.renderVideo` reports frame counts when `_meta.progressToken` is present (a string or number). Notifications increase strictly,