pauljd/photocraft
1
0
Fork 0
mirror of https://github.com/storytold/photocraft.git synced 2026-11-07 12:13:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

MCP bridge: waited commands poll the app's job, with progress and cancel (#8130) (#8134)

Browse Source 
* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>
Co-authored-by: Khawar Zamman Wani <wanikhawar@Khawars-MacBook-Air.local>
Co-authored-by: FredApps <fredapps@users.noreply.github.com>
This commit is contained in:
Khawar Zamman Wani
2026-11-05 13:35:13 +01:00
committed by GitHub
co-authored by Claude Opus 5.5 Khawar Zamman Wani FredApps
parent 888e87d606
commit 274f6d3e26
9 changed files with 584 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
book/src/automation/mcp.md
+4 -3
View File
@@ -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.
crates/automation/src/bridge.rs
+15 -25
View File
@@ -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(())),
}
}
}
crates/automation/src/server.rs
+113 -62
View File
@@ -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<RunParams>, context: RequestContext<RoleServer>) -> Result<CallToolResult, McpError> {
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;
crates/automation/src/server/progress.rs
+155 -12
View File
@@ -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<Mutex<Headless>>,
backend: Arc<Backend>,
job: Option<JobId>,
}
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<Result<Value, AutomationError>> {
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<Value, AutomationError> {
// 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<Value, AutomationError>) -> 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}");
}
}
crates/automation/tests/it/bridge.rs
+36
View File
@@ -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::<Value>(&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");
}
crates/automation/tests/it/mcp.rs
+8 -4
View File
@@ -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),
crates/automation/tests/mcp_wait.rs
+235 -10
View File
@@ -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<u64>) -> 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<u64>) -> (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<Instant>,
cancelled: bool,
log: Vec<Value>,
}
impl FakeApp {
async fn serve(runs: Duration, lose_start: bool) -> (String, Arc<Mutex<FakeApp>>) {
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<FakeApp>, method: &str) -> Vec<Value> {
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;
docs/control-protocol.md
+2 -2
View File
@@ -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 <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. 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:<port>`), 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}` |
docs/mcp.md
+16 -12
View File
@@ -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,