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 77c4b3d4fc..4f210b07f4 100644 --- a/crates/automation/src/server.rs +++ b/crates/automation/src/server.rs @@ -989,8 +989,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) @@ -1096,7 +1096,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, } } @@ -1388,65 +1389,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")] @@ -1516,6 +1458,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/crates/compose/src/cmyk.rs b/crates/compose/src/cmyk.rs index c09ccd4fd1..acf41639bd 100644 --- a/crates/compose/src/cmyk.rs +++ b/crates/compose/src/cmyk.rs @@ -417,6 +417,10 @@ fn mix_adjusted(l: &Layer, base: &mut CmykBuffer, adjusted: &CmykBuffer, cx: &Ct let opacity = l.opacity * l.fill_opacity; let special = fill_special && l.fill_opacity < 1.0 && super::psblend::fill_is_special(l.blend); let fill = l.fill_opacity.clamp(0.0, 1.0); + if special && fill <= 0.0 { + // Zero Fill is the backdrop: keep its inks without a brightness round trip. + return; + } let coverage = if special { l.opacity } else { opacity }; let rect = base.rect; let w = (rect.width() as usize).max(1); diff --git a/crates/compose/src/cmyk/tests.rs b/crates/compose/src/cmyk/tests.rs index 1118250132..8deaa68de3 100644 --- a/crates/compose/src/cmyk/tests.rs +++ b/crates/compose/src/cmyk/tests.rs @@ -2008,3 +2008,49 @@ fn adjustment_layers_keep_the_special_fill_rule() { assert_routes_agree(&d); } } + +/// A one-pixel 32-bit CMYK document: a raster `base` backdrop and an Invert adjustment layer in +/// `blend` at `fill`; `clipped`, when given, is a fully covering clipped Normal raster that +/// replaces the adjusted inks exactly. +fn special_fill_doc(base: Cmyka, blend: BlendMode, fill: f32, clipped: Option) -> Document { + let format = PixelFormat { mode: ColorMode::Cmyk, sample: SampleType::F32, alpha: true }; + let pixel = Rect::new(0, 0, 1, 1); + let mut d = Document::new("fill", Size::new(1, 1), ColorMode::Cmyk, SampleType::F32); + let mut backdrop = Layer::raster("base", format); + backdrop.surface_mut().unwrap().fill_rect(pixel, &base); + d.layers.push(backdrop); + let mut adjust = adjustment(Adjustment::Invert); + adjust.blend = blend; + adjust.fill_opacity = fill; + d.layers.push(adjust); + if let Some(ink) = clipped { + let mut clip = Layer::raster("clip", format); + clip.surface_mut().unwrap().fill_rect(pixel, &ink); + clip.clipped = true; + d.layers.push(clip); + } + d +} + +/// Zero Fill leaves the backdrop bit-exact in all of the special eight. Linear Burn and Linear +/// Light used to round trip through brightness (0.7 came back as 0.70000005). +#[test] +fn adjustment_zero_fill_keeps_backdrop_inks_exact() { + let base = [0.1, 0.4, 0.7, 0.3, 1.0]; + for mode in SEPARABLE.iter().copied().filter(|m| crate::psblend::fill_is_special(*m)) { + let d = special_fill_doc(base, mode, 0.0, None); + assert!(supports(&d)); + assert_eq!(render1(&d, 0, 0), base, "{mode:?}"); + } +} + +/// Two F32 inks can share one brightness; zero Fill keeps the backdrop's, not the source's. +#[test] +fn adjustment_zero_fill_preserves_colliding_brightness() { + let base = [f32::from_bits(0x3dcccccc), 0.4, 0.7, 0.3, 1.0]; + let source = [f32::from_bits(0x3dcccccd), 0.4, 0.7, 0.3, 1.0]; + assert_ne!(base[0], source[0]); + assert_eq!(1.0 - base[0], 1.0 - source[0]); + let d = special_fill_doc(base, BlendMode::Difference, 0.0, Some(source)); + assert_eq!(render1(&d, 0, 0), base); +} diff --git a/crates/io/src/flat.rs b/crates/io/src/flat.rs index 64e6ee7fce..101071eee7 100644 --- a/crates/io/src/flat.rs +++ b/crates/io/src/flat.rs @@ -574,7 +574,7 @@ fn document_format(layout: ChannelLayout, sample: CSample) -> DocumentFormat { /// A decoded flat image as a single-layer document: a locked "Background" when it is opaque, a /// normal "Layer 0" when it has transparency. -pub(crate) fn image_to_document(name: &str, img: &Image) -> Result { +pub fn image_to_document(name: &str, img: &Image) -> Result { image_to_document_with(name, img, &Interrupt::NONE) } diff --git a/crates/io/src/lib.rs b/crates/io/src/lib.rs index e5dd351c89..6cb0aafc84 100644 --- a/crates/io/src/lib.rs +++ b/crates/io/src/lib.rs @@ -97,6 +97,9 @@ pub use flat::{ export_png_streamed_with_renderer, export_preview_png_with_renderer, file_encoding, import_gif_frames, import_tiff_page, import_tiff_page_with_options, import_webp_frames, tga_alpha_channel_to_transparency, tga_default_bits, }; +// Re-exported for the `flat_open_bench` example, which times the tile-building phase of a flat open. +#[doc(hidden)] +pub use flat::image_to_document; pub use psd_export::{PsdExportOptions, document_to_psd, document_to_psd_with}; pub use psd_import::{psd_to_document, psd_to_document_with}; diff --git a/crates/ui-egui/Cargo.toml b/crates/ui-egui/Cargo.toml index e046f69001..0af8e7ac17 100644 --- a/crates/ui-egui/Cargo.toml +++ b/crates/ui-egui/Cargo.toml @@ -81,3 +81,7 @@ flate2 = { workspace = true } epaint_default_fonts = { version = "0.37", features = ["monochrome_emoji_fonts"] } serde_json = { workspace = true } toml = "1" + +[[example]] +name = "flat_open_bench" +test = true diff --git a/crates/ui-egui/examples/flat_open_bench.rs b/crates/ui-egui/examples/flat_open_bench.rs new file mode 100644 index 0000000000..447291acaf --- /dev/null +++ b/crates/ui-egui/examples/flat_open_bench.rs @@ -0,0 +1,449 @@ +//! Flat-image open benchmark (#10259): File › Open of an ordinary JPEG or PNG, phase by phase, +//! with the peak resident memory of each case. +//! +//! ```sh +//! cargo run --release -p photocraft-ui-egui --example flat_open_bench -- [--quick] [--reps 5] [--only jpeg-36] [--cpu] [--cache DIR] [--json out.json] +//! cargo run --release -p photocraft-ui-egui --example flat_open_bench -- --case png16-36 # one case, one process +//! ``` +//! +//! Cases (`--only` matches a substring of the key): `jpeg-{24,36}` (RGB, quality 90, 4:2:0), +//! `png8-{24,36}` (RGB 8-bit), `png16-{24,36}` (RGB 16-bit), `rgba8-{24,36}` (RGB 8-bit with +//! partial alpha) and `jpeg-24-orient6` (EXIF orientation 6). 24 MP is 6000x4000, 36 MP is +//! 7360x4912; `--quick` uses 1200x800 for every case (the row names keep the nominal size). +//! +//! The inputs are generated in-process (smooth gradients, soft blobs and seeded noise, so the +//! compression is realistic) with the workspace's own encoders and cached in `--cache` (default: +//! `$PHOTOCRAFT_FIXTURE_DIR`, else `fixtures/` in the target directory), never committed. +//! +//! Rows per case, in ms over `--reps` repetitions: +//! - `import`: `import_for_open_with`, the real File › Open path (decode, orientation, tiles) +//! - `admission`: `Session::open_document` of that document +//! - `decode`: the codec alone, orientation kept as stored +//! - `orient`: `Image::oriented` with the file's EXIF orientation (a no-op for an upright file) +//! - `tiles`: `image_to_document` on the decoded image (layout conversion, tile construction, prune) +//! - `first composite`: the canvas's first full refresh (GPU, or the CPU compositor with `--cpu` +//! or without an adapter), as `perf_scenarios` times it +//! - `editable`: import + admission + first composite of one open, the "decode + first composite" rows +//! +//! Peak memory: without `--case` the bench starts itself once per case (`current_exe`), so each +//! case's `VmHWM` is its own process-lifetime high-water mark. `peak_process_after_import_bytes` +//! is read after real import and admission; `peak_process_after_all_passes_bytes` is read after +//! standalone decode/orient/tiles, GPU setup and warm-up, and editable-open passes. The latter +//! is not an isolated composite or editable-open peak. Row `peak_rss_provenance` identifies +//! which reading its `peak_rss_bytes` carries. Off Linux the peak is `n/a`. +//! A case run directly with `--case` generates a missing fixture in its own process, which +//! raises its peak: run the bench once without `--case` first. + +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::Instant; + +use eframe::egui_wgpu::RenderState; +use photocraft_codecs::{ChannelLayout, DecodeOptions, EncodeOptions, Format, Image}; +use photocraft_engine::Session; +use photocraft_ui_egui::gpu_canvas::GpuCanvas; +use rayon::prelude::*; +use serde_json::{Value, json}; + +/// Prefix of the line a `--case` process prints for its parent. +const RESULT_PREFIX: &str = "FLAT_OPEN_RESULT "; +/// The phases, in report order. +const PHASES: [&str; 7] = ["import", "admission", "decode", "orient", "tiles", "first composite", "editable"]; + +#[derive(Clone, Copy, PartialEq)] +enum Kind { + Jpeg, + Png8, + Png16, + Rgba8, +} + +struct Case { + key: &'static str, + label: &'static str, + kind: Kind, + /// 36 MP (7360x4912) instead of 24 MP (6000x4000). + big: bool, + /// EXIF orientation 6 (the stored image is turned upright on open). + orient6: bool, +} + +const CASES: [Case; 9] = [ + Case { key: "jpeg-24", label: "open 24 MP JPEG RGB8", kind: Kind::Jpeg, big: false, orient6: false }, + Case { key: "jpeg-36", label: "open 36 MP JPEG RGB8", kind: Kind::Jpeg, big: true, orient6: false }, + Case { key: "png8-24", label: "open 24 MP PNG RGB8", kind: Kind::Png8, big: false, orient6: false }, + Case { key: "png8-36", label: "open 36 MP PNG RGB8", kind: Kind::Png8, big: true, orient6: false }, + Case { key: "png16-24", label: "open 24 MP PNG RGB16", kind: Kind::Png16, big: false, orient6: false }, + Case { key: "png16-36", label: "open 36 MP PNG RGB16", kind: Kind::Png16, big: true, orient6: false }, + Case { key: "rgba8-24", label: "open 24 MP PNG RGBA8", kind: Kind::Rgba8, big: false, orient6: false }, + Case { key: "rgba8-36", label: "open 36 MP PNG RGBA8", kind: Kind::Rgba8, big: true, orient6: false }, + Case { key: "jpeg-24-orient6", label: "open 24 MP JPEG RGB8 (EXIF orientation 6)", kind: Kind::Jpeg, big: false, orient6: true }, +]; + +impl Case { + fn size(&self, quick: bool) -> (u32, u32) { + match (quick, self.big) { + (true, _) => (1200, 800), + (false, false) => (6000, 4000), + (false, true) => (7360, 4912), + } + } + fn ext(&self) -> &'static str { + if self.kind == Kind::Jpeg { "jpg" } else { "png" } + } + fn file_name(&self, w: u32, h: u32) -> String { + format!("flat_open_{}_{w}x{h}.{}", self.key, self.ext()) + } +} + +/// Return the value following a named command-line option. +fn arg(args: &[String], name: &str) -> Option { + args.iter().position(|a| a == name).and_then(|i| args.get(i + 1).cloned()) +} + +fn ms(t: Instant) -> f64 { + t.elapsed().as_secs_f64() * 1000.0 +} + +/// `/fixtures`: the binary lives in `/release/examples/`. +fn default_cache() -> PathBuf { + if let Some(d) = std::env::var_os("PHOTOCRAFT_FIXTURE_DIR") { + return PathBuf::from(d); + } + std::env::current_exe().ok().and_then(|e| e.ancestors().nth(3).map(|t| t.join("fixtures"))).unwrap_or_else(std::env::temp_dir) +} + +/// Deterministic noise in [-0.5, 0.5) from the pixel position and a channel/seed. +fn noise(x: usize, y: usize, seed: usize) -> f32 { + let mut h = (x.wrapping_mul(73_856_093) ^ y.wrapping_mul(19_349_663) ^ seed.wrapping_mul(83_492_791)) as u64; + h = (h ^ (h >> 15)).wrapping_mul(0x2c1b_3c6d); + h = (h ^ (h >> 12)).wrapping_mul(0x297a_2d39); + ((h ^ (h >> 15)) % 4096) as f32 / 4096.0 - 0.5 +} + +/// A photo-like sample in [0, 1] for channel `c` (3 is alpha: opaque at the centre, down to +/// about 15 % at the corners, with fine noise, so most pixels are partly transparent). +fn sample(x: usize, y: usize, w: usize, h: usize, c: usize) -> f32 { + let (fx, fy) = (x as f32 / w as f32, y as f32 / h as f32); + let n = noise(x, y, c); + let blob = (-(((fx - 0.4) * 3.0).powi(2) + ((fy - 0.55) * 4.0).powi(2))).exp(); + let v = match c { + 0 => 0.55 + 0.17 * (fx * 6.0).sin() + 0.3 * blob + n * 0.06, + 1 => 0.45 + 0.12 * (fy * 9.0 + fx * 2.0).cos() + 0.2 * blob + n * 0.06, + 2 => 0.35 + 0.4 * fy - 0.2 * blob + n * 0.06, + _ => 1.0 - 0.85 * (((fx - 0.5).powi(2) + (fy - 0.5).powi(2)) * 2.0).min(1.0) + n * 0.04, + }; + v.clamp(0.0, 1.0) +} + +/// A TIFF-structured EXIF block holding only the orientation tag. +fn exif_orientation_block(o: u16) -> Vec { + let mut b = b"II*\0\x08\0\0\0\x01\0".to_vec(); + b.extend_from_slice(&[0x12, 0x01, 3, 0, 1, 0, 0, 0]); + b.extend_from_slice(&o.to_le_bytes()); + b.extend_from_slice(&[0, 0, 0, 0, 0, 0]); + b +} + +/// The encoded file for `case` at `w`x`h`. +fn generate(case: &Case, w: u32, h: u32) -> Result, String> { + let (wu, hu) = (w as usize, h as usize); + let (layout, ch) = if case.kind == Kind::Rgba8 { (ChannelLayout::Rgba, 4) } else { (ChannelLayout::Rgb, 3) }; + let img = if case.kind == Kind::Png16 { + let mut px = vec![0u16; wu * hu * ch]; + px.par_chunks_mut(wu * ch).enumerate().for_each(|(y, row)| { + for (i, v) in row.iter_mut().enumerate() { + *v = (sample(i / ch, y, wu, hu, i % ch) * 65535.0 + 0.5) as u16; + } + }); + Image::from_u16(w, h, layout, &px).map_err(|e| e.to_string())? + } else { + let mut px = vec![0u8; wu * hu * ch]; + px.par_chunks_mut(wu * ch).enumerate().for_each(|(y, row)| { + for (i, v) in row.iter_mut().enumerate() { + *v = (sample(i / ch, y, wu, hu, i % ch) * 255.0 + 0.5) as u8; + } + }); + Image::from_u8(w, h, layout, px).map_err(|e| e.to_string())? + }; + let format = if case.kind == Kind::Jpeg { Format::Jpeg } else { Format::Png }; + let opts = EncodeOptions { jpeg_quality: 90, jpeg_chroma_subsampling: true, ..Default::default() }; + let mut bytes = photocraft_codecs::encode(&img, format, &opts).map_err(|e| e.to_string())?; + if case.orient6 { + // The encoder writes the pixels as given and resets the orientation, so the tag goes in + // as an Exif APP1 segment right after the JPEG's SOI marker. + let exif = exif_orientation_block(6); + let len = u16::try_from(exif.len() + 8).map_err(|e| e.to_string())?; + let mut seg = vec![0xFF, 0xE1]; + seg.extend_from_slice(&len.to_be_bytes()); + seg.extend_from_slice(b"Exif\0\0"); + seg.extend_from_slice(&exif); + bytes.splice(2..2, seg); + } + Ok(bytes) +} + +/// The cached fixture for `case`, generated (and written atomically) when missing. +fn ensure_fixture(case: &Case, dir: &Path, quick: bool) -> Result { + let (w, h) = case.size(quick); + let path = dir.join(case.file_name(w, h)); + if path.metadata().is_ok_and(|m| m.len() > 0) { + return Ok(path); + } + std::fs::create_dir_all(dir).map_err(|e| format!("create {}: {e}", dir.display()))?; + let t = Instant::now(); + let bytes = generate(case, w, h).map_err(|e| format!("generate {}: {e}", case.key))?; + let tmp = path.with_extension(format!("{}.tmp{}", case.ext(), std::process::id())); + std::fs::write(&tmp, &bytes).map_err(|e| format!("write {}: {e}", tmp.display()))?; + std::fs::rename(&tmp, &path).map_err(|e| format!("rename {}: {e}", path.display()))?; + eprintln!("generated {} ({:.1} MB) in {:.1} s", path.display(), bytes.len() as f64 / 1e6, t.elapsed().as_secs_f64()); + Ok(path) +} + +/// A headless GPU canvas on a device created like the app's. +fn canvas() -> Option<(GpuCanvas, RenderState)> { + let setup = photocraft_ui_egui::gpu_canvas::wgpu_setup(); + let rs = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| egui_kittest::wgpu::create_render_state(setup, Default::default()))).ok()?; + Some((GpuCanvas::new(&rs), rs)) +} + +/// The canvas's first full refresh; waits for the GPU like presenting the next frame would. +fn refresh(gpu: Option<&(GpuCanvas, RenderState)>, s: &Session) -> Result { + let st = s.active().ok_or("no document")?; + let t = Instant::now(); + match gpu { + Some((g, rs)) => { + g.refresh(st.doc.id.0, &st.doc, None, None); + let _ = rs.device.poll(eframe::wgpu::PollType::Wait { submission_index: None, timeout: None }); + } + None => { + std::hint::black_box(photocraft_compose::render(&st.doc, st.doc.bounds())); + } + } + Ok(ms(t)) +} + +/// `import_for_open_with`, File › Open's real path. +fn import_once(name: &str, bytes: &[u8]) -> Result { + let opts = photocraft_io::ImportOptions::default(); + photocraft_io::import_for_open_with(name, bytes, &opts, &Default::default()).map(|(r, _)| r.document).map_err(|e| e.to_string()) +} + +/// Runs every measurement for one case in this process; the result is the JSON line's payload. +fn run_case(case: &Case, dir: &Path, quick: bool, reps: usize, cpu: bool) -> Result { + let (w, h) = case.size(quick); + let path = ensure_fixture(case, dir, quick)?; + let bytes = std::fs::read(&path).map_err(|e| format!("read {}: {e}", path.display()))?; + let name = format!("photo.{}", case.ext()); + let want = if case.orient6 { (h, w) } else { (w, h) }; + let mut ph: Vec> = vec![Vec::new(); PHASES.len()]; + let mut put = |i: usize, v: f64| { + if let Some(p) = ph.get_mut(i) { + p.push(v); + } + }; + + // Pass 1: the real path and the admission, before anything else raises the process's peak. + for _ in 0..reps { + let t = Instant::now(); + let doc = import_once(&name, &bytes)?; + put(0, ms(t)); + if doc.size.width != want.0 || doc.size.height != want.1 { + return Err(format!("{}: opened {}x{}, expected {}x{}", case.key, doc.size.width, doc.size.height, want.0, want.1)); + } + let t = Instant::now(); + let mut s = Session::new(); + s.open_document(doc, Some(name.clone())); + put(1, ms(t)); + } + let peak_process_after_import = photocraft_testkit::perf::process_peak_rss_bytes(); + + // Pass 2: the phases of the import on their own. + let format = photocraft_codecs::detect(&bytes).ok_or("unknown format")?; + let keep = DecodeOptions { keep_orientation: true, ..Default::default() }; + for _ in 0..reps { + let t = Instant::now(); + let img = photocraft_codecs::decode_as_with(format, &bytes, &keep).map_err(|e| e.to_string())?; + put(2, ms(t)); + let o = photocraft_codecs::exif_orientation(img.meta.exif.as_deref().unwrap_or_default()); + let t = Instant::now(); + let upright = img.oriented(o).map_err(|e| e.to_string())?; + put(3, ms(t)); + let t = Instant::now(); + let r = photocraft_io::image_to_document(&name, &upright).map_err(|e| e.to_string())?; + put(4, ms(t)); + drop(r); + } + + // Pass 3: one whole open as perf_scenarios times it: import, admission, first composite. + let gpu = if cpu { None } else { canvas() }; + let adapter = gpu.as_ref().map(|(_, rs)| rs.adapter.get_info().name); + // One untimed open first, so the GPU pipeline and adapter warm-up is not in the first sample. + { + let mut s = Session::new(); + s.open_document(import_once(&name, &bytes)?, Some(name.clone())); + refresh(gpu.as_ref(), &s)?; + } + for _ in 0..reps { + let t = Instant::now(); + let doc = import_once(&name, &bytes)?; + let mut s = Session::new(); + s.open_document(doc, Some(name.clone())); + let open = ms(t); + let c = refresh(gpu.as_ref(), &s)?; + put(5, c); + put(6, open + c); + } + let peak_process_after_all_passes = photocraft_testkit::perf::process_peak_rss_bytes(); + let phases: serde_json::Map = PHASES.iter().zip(ph).map(|(n, v)| ((*n).to_string(), json!(v))).collect(); + Ok(json!({ + "key": case.key, "label": case.label, "width": w, "height": h, "file_bytes": bytes.len(), + "peak_process_after_import_bytes": peak_process_after_import, "peak_process_after_all_passes_bytes": peak_process_after_all_passes, "adapter": adapter, "phases": phases, + })) +} + +/// The report rows of one case's result: `