Merge remote-tracking branch 'origin/main' into integration-batch-17

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.

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(())),
         }
     }
 }

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<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)
@@ -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;

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<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

@@ -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

@@ -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

@@ -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;

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);

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<Cmyka>) -> 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);
+}

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<ImportResult, IoError> {
+pub fn image_to_document(name: &str, img: &Image) -> Result<ImportResult, IoError> {
     image_to_document_with(name, img, &Interrupt::NONE)
 }
 

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};
 

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

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<String> {
+    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
+}
+
+/// `<target>/fixtures`: the binary lives in `<target>/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<u8> {
+    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<Vec<u8>, 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<PathBuf, String> {
+    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<f64, String> {
+    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<photocraft_doc::Document, String> {
+    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<Value, String> {
+    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<f64>> = 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<String, Value> = 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: `<label>: <phase>`.
+fn rows_of(res: &Value) -> Vec<Value> {
+    let label = res["label"].as_str().unwrap_or("?");
+    PHASES
+        .iter()
+        .map(|p| {
+            let samples: Vec<f64> = res["phases"][*p].as_array().map(|a| a.iter().filter_map(Value::as_f64).collect()).unwrap_or_default();
+            let composite = matches!(*p, "first composite" | "editable");
+            let peak = res[if composite { "peak_process_after_all_passes_bytes" } else { "peak_process_after_import_bytes" }].as_u64();
+            let mut row = photocraft_testkit::perf::row(&format!("{label}: {p}"), &samples, peak, None);
+            row["peak_rss_provenance"] = json!(if composite { "whole_case_process_after_all_passes" } else { "case_process_after_import_and_admission" });
+            row
+        })
+        .collect()
+}
+
+/// Format a measured byte count in MiB, retaining unavailable measurements.
+fn mb(v: &Value) -> String {
+    v.as_u64().map_or_else(|| "n/a".into(), |b| format!("{}", b >> 20))
+}
+
+/// Print phase timings and identify the scope of both process RSS readings.
+fn print_case(res: &Value) {
+    println!("{} ({}x{}, {:.1} MB file)", res["label"].as_str().unwrap_or("?"), res["width"], res["height"], res["file_bytes"].as_f64().unwrap_or(0.0) / 1e6);
+    for r in rows_of(res) {
+        let (name, p50, p95) = (r["name"].as_str().unwrap_or("?"), r["p50_ms"].as_f64(), r["p95_ms"].as_f64());
+        let phase = name.rsplit(": ").next().unwrap_or(name);
+        println!("  {phase:<16} p50 {:>8.1} ms   p95 {:>8.1} ms", p50.unwrap_or(f64::NAN), p95.unwrap_or(f64::NAN));
+    }
+    println!(
+        "  process RSS high-water: {} MiB after import/admission, {} MiB after all passes (including standalone phases and warm-up)",
+        mb(&res["peak_process_after_import_bytes"]),
+        mb(&res["peak_process_after_all_passes_bytes"])
+    );
+}
+
+/// Starts this binary for one case so its peak memory is its own; returns the parsed result.
+fn spawn_case(case: &Case, args: &[String], dir: &Path, reps: usize) -> Result<Value, String> {
+    let exe = std::env::current_exe().map_err(|e| format!("current_exe: {e}"))?;
+    let mut cmd = Command::new(exe);
+    cmd.args(["--case", case.key, "--reps", &reps.to_string(), "--cache"]).arg(dir).stdout(Stdio::piped()).stderr(Stdio::inherit());
+    for flag in ["--quick", "--cpu"] {
+        if args.iter().any(|a| a == flag) {
+            cmd.arg(flag);
+        }
+    }
+    let out = cmd.output().map_err(|e| format!("start {}: {e}", case.key))?;
+    if !out.status.success() {
+        return Err(format!("{}: the case process failed ({})", case.key, out.status));
+    }
+    let text = String::from_utf8_lossy(&out.stdout);
+    let line = text.lines().find_map(|l| l.strip_prefix(RESULT_PREFIX)).ok_or_else(|| format!("{}: no result line", case.key))?;
+    serde_json::from_str(line).map_err(|e| format!("{}: bad result: {e}", case.key))
+}
+
+/// Run isolated benchmark cases and optionally write their combined report.
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    let quick = args.iter().any(|a| a == "--quick");
+    let cpu = args.iter().any(|a| a == "--cpu");
+    let reps: usize = arg(&args, "--reps").and_then(|v| v.parse().ok()).unwrap_or(if quick { 3 } else { 5 }).max(1);
+    let dir = arg(&args, "--cache").map_or_else(default_cache, PathBuf::from);
+    let only = arg(&args, "--only");
+    let mut rows = Vec::new();
+    let mut results = Vec::new();
+    let mut errors = Vec::new();
+
+    if let Some(key) = arg(&args, "--case") {
+        // One case in this process: print the result line for the parent (or a person).
+        let Some(case) = CASES.iter().find(|c| c.key == key) else {
+            eprintln!("unknown --case {key} (one of: {})", CASES.iter().map(|c| c.key).collect::<Vec<_>>().join(", "));
+            std::process::exit(2);
+        };
+        match run_case(case, &dir, quick, reps, cpu) {
+            Ok(res) => {
+                print_case(&res);
+                println!("{RESULT_PREFIX}{res}");
+                rows.extend(rows_of(&res));
+                results.push(res);
+            }
+            Err(e) => {
+                eprintln!("{e}");
+                errors.push(json!({"name": key, "error": e}));
+            }
+        }
+    } else {
+        for case in CASES.iter().filter(|c| only.as_deref().is_none_or(|o| c.key.contains(o))) {
+            // Generated here, so the case process only reads it.
+            let res = ensure_fixture(case, &dir, quick).and_then(|_| spawn_case(case, &args, &dir, reps));
+            match res {
+                Ok(res) => {
+                    print_case(&res);
+                    rows.extend(rows_of(&res));
+                    results.push(res);
+                }
+                Err(e) => {
+                    eprintln!("{e}");
+                    errors.push(json!({"name": case.key, "error": e}));
+                }
+            }
+        }
+    }
+
+    if let Some(out) = arg(&args, "--json") {
+        let cases: Vec<Value> = results
+            .iter()
+            .map(|r| json!({"key": r["key"], "width": r["width"], "height": r["height"], "file_bytes": r["file_bytes"], "peak_process_after_import_bytes": r["peak_process_after_import_bytes"], "peak_process_after_all_passes_bytes": r["peak_process_after_all_passes_bytes"]}))
+            .collect();
+        let adapter = results.iter().find_map(|r| r["adapter"].as_str().map(str::to_string));
+        let context = json!({
+            "reps": reps, "quick": quick, "cache": dir.display().to_string(), "cases": cases, "errors": errors, "gpu_adapter": adapter,
+            "threads": std::thread::available_parallelism().map(|n| n.get()).ok(),
+        });
+        let mut report = photocraft_testkit::perf::report("flat_open_bench", context, rows, None);
+        // Each case ran in its own process: this one's peak says nothing about them.
+        report["process_peak_rss_bytes"] = json!(results.iter().filter_map(|r| r["peak_process_after_all_passes_bytes"].as_u64()).max());
+        if let Err(e) = photocraft_testkit::perf::write_report(&out, &report) {
+            eprintln!("{e}");
+            errors.push(json!({"name": "report", "error": e}));
+        }
+    }
+    if !errors.is_empty() {
+        std::process::exit(1);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rows_identify_the_process_peak_reading() {
+        let res = json!({
+            "label": "fixture", "phases": {},
+            "peak_process_after_import_bytes": 100,
+            "peak_process_after_all_passes_bytes": 200,
+        });
+        let rows = rows_of(&res);
+        for (row, phase) in rows.iter().zip(PHASES) {
+            let all_passes = matches!(phase, "first composite" | "editable");
+            assert_eq!(row["peak_rss_bytes"], if all_passes { 200 } else { 100 });
+            assert_eq!(row["peak_rss_provenance"], if all_passes { "whole_case_process_after_all_passes" } else { "case_process_after_import_and_admission" });
+        }
+    }
+}

docs/control-protocol.md

@@ -362,7 +362,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 \
@@ -373,7 +373,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/development.md

@@ -440,6 +440,7 @@ can't run.
 - **Numbers** (7360 × 4912, 8 text layers + one painted layer with drop shadow + stroke + bevel, Hue/Saturation on top; M4 Pro; `cargo run --release -p photocraft-ui-egui --example fx_bench`), CPU fallback → GPU: full refresh with warm effect maps 5.7 s → 47 ms; Hue/Saturation tweak above the effects 4.7 s → 31 ms; brush dab on a plain layer 32 → 1.7 ms; brush dab on the effect layer (its maps rebuilt around the dab) 659 → 3.7 ms; first refresh (all maps built) 6.4 s → 0.39 s. With the CPU ~14× oversubscribed by parallel builds (min of 7 runs): 8.1 s → 0.33 s, 9.9 s → 0.28 s, 28 → 2.2 ms, 1.7 s → 89 ms; moving a text layer 7 px 336 → 13 ms. `fx_bench --compare corpus/psd/…/*.psd` reports the GPU vs CPU difference on real files (30 of the 31 corpus files with effects render on the GPU, worst 0.12/255).
 - On the CPU path `compose::effect_maps` caches shadow, glow, bevel and satin maps per layer state (LRU, 768 MB budget).
 - Live adjustment previews on large documents use a downsampled proxy (`ui-egui/src/proxy.rs`).
+- **Opening JPEG and PNG files** (#10259). `cargo run --release -p photocraft-ui-egui --example flat_open_bench` generates 24 and 36 MP JPEG, PNG RGB 8-bit, RGB 16-bit and RGBA 8-bit inputs (and a JPEG with EXIF orientation 6) from seeded gradients and noise, caches them in the target directory's `fixtures/` (`--cache`), and times the real open (`import`), session `admission`, the codec `decode`, `orient`, `tiles` (`image_to_document`), the `first composite` and the whole `editable` chain. Each case runs in its own process (`--case <key>` runs one by hand). Its peak resident memory (`VmHWM`, Linux only, else `n/a`) is a process-lifetime high-water mark: `peak_process_after_import_bytes` is read after real import and admission, while `peak_process_after_all_passes_bytes` also includes the standalone decode/orient/tiles pass, GPU setup and warm-up, and editable-open passes. The latter is not an isolated composite or editable-open peak. Each timing row identifies which reading it carries in `peak_rss_provenance`. `--quick` uses 1200x800 images, `--only <substring>` filters cases, `--cpu` skips the GPU, `--json <file>` writes the report `cargo xtask perf` reads (scenarios P83 to P86).
 - **Content-Aware tools** (`algo::inpaint`, `algo::content_aware`: Fill, Spot Healing, Patch, Content-Aware Move). `cargo run --release -p photocraft-engine --example content_aware_bench -- bench caf|heal|cancel` times the P21, P22 and P49 rows without the UI, one sample per line, so two builds can be run in turn and pooled. `-- hash` prints a hash of every tool's output at 8, 16 and 32 bits and in Gray, CMYK and Lab: the fill is fitted to Photoshop, so a speed-up must leave both builds' lists equal. The solver works only over the hole's bounds grown by a patch radius (nothing else changes or is matched). PatchMatch sweeps its bands in wavefronts when there are fewer bands than threads; each cell sees the same neighbours and random numbers as in raster order, so the result does not depend on the thread count. Cancellation is checked per row: a 24 MP fill stops within tens of milliseconds, most of it spent freeing its buffers.
 - `ui.inspect` returns `perf` timings (UI ms per frame, composite ms, upload ms).
 - Never scan full surfaces per frame. Cache per document revision (`PhotocraftApp::cached_bounds`). An uncached `content_bounds()` on a 36 MP layer once cost 77 ms per frame.

docs/mcp.md

@@ -137,18 +137,22 @@ for ten minutes. List/read responses include the MCP 2026-07-28 result/cache fie
 
 ## Progress and cancellation
 
-Headless `command_run` (and each `command_batch` step) that waits, the default, runs a
-job-capable command (filters, Content-Aware Fill/Scale, Photomerge, …) as a background job
-and returns its result when the job ends. Meanwhile the session serves other calls
-(`doc_inspect`, `jobs_list`, …), and the running job shows in `jobs_list`. With
-`_meta.progressToken`, the call sends `notifications/progress` with the job's progress (0–1,
-`total` 1; for a batch, finished steps plus the current step's fraction, `total` the step
-count) and the job's `message`, increasing strictly and at most ten per second.
-`notifications/cancelled`, or the client dropping the request, cancels the job: the document
-stays unchanged and a cancelled batch runs no further step. A job that already finished
-before the cancellation arrived stays applied. Commands that are not jobs still run to the
-end. Bridge mode still waits on the app's reply (#8130); a bridge `command_batch` reports
-finished steps as progress and sends no further step once the request is cancelled.
+`command_run` (and each `command_batch` step) that waits, the default, runs a job-capable
+command (filters, Content-Aware Fill/Scale, Photomerge, …) as a background job and returns its
+result when the job ends. Meanwhile the server serves other calls (`doc_inspect`, `jobs_list`,
+…), and the running job shows in `jobs_list`. With `_meta.progressToken`, the call sends
+`notifications/progress` with the job's progress (0–1, `total` 1; for a batch, finished steps
+plus the current step's fraction, `total` the step count) and the job's `message`, increasing
+strictly and at most ten per second. `notifications/cancelled`, or the client dropping the
+request, cancels the job: the document stays unchanged and a cancelled batch runs no further
+step. A job that already finished before the cancellation arrived stays applied. Commands that
+are not jobs still run to the end.
+
+Headless, the session runs the job itself (#8129). In bridge mode (#8130) the app runs it: the
+server starts it with `engine.execute {wait: false}` (sent once, never resent) and polls
+`jobs.list`, so a job may run past the bridge's 55s reply timeout. A command that is not a job,
+a UI command id and a lost reply to the start still fail at that timeout; see
+[MCP bridge](control-protocol.md#mcp-bridge) for the limits.
 
 A direct headless `command_run` of `file.export.renderVideo` reports frame counts when
 `_meta.progressToken` is present (a string or number). Notifications increase strictly,

docs/scorecard.md

@@ -6,7 +6,7 @@ Generated by `cargo xtask scorecard` from `perf/budgets.toml`, `perf/baseline.js
 
 | Area | Done | Partial | Missing | Key numbers |
 |---|---:|---:|---:|---|
-| [Performance](#performance) | 3 | 22 | 45 | 79 scenarios: 3 within budget, 22 over budget or failing, 9 tracked only, 41 not run, 4 not measurable yet (baseline 2026-10-05) |
+| [Performance](#performance) | 3 | 22 | 49 | 83 scenarios: 3 within budget, 22 over budget or failing, 9 tracked only, 45 not run, 4 not measurable yet (baseline 2026-10-05) |
 | [File compatibility](#file-compatibility) | 11 | 14 | 2 | io corpus: oracle ≥ 148/170, round trip ≥ 169/170; psd-tools corpus: oracle ≥ 249/309, round trip ≥ 308/309; photoshop oracles: oracle ≥ 190/258, round trip ≥ 258/258 |
 | [Tools](#tools) | 36 | 9 | 0 |  |
 | [Workspace and UI](#workspace-and-ui) | 15 | 6 | 2 | settings that do nothing: 10 of 189 (target 0) |
@@ -101,6 +101,10 @@ Baseline: 2026-10-05, 1 run, commit `f5d2bcb54a`, Apple M4 Pro (48 GB), GPU Appl
 | P80 | App frame: full CPU refresh of a 24 MP document | ≤ 50 ms p95 | – | – | – | – | – | not run | [#9328](https://github.com/storytold/photocraft/issues/9328) |
 | P81 | Undo a brush dab on a full-canvas 16-bit 15000 x 10000 layer + refresh | track only | – | – | – | – | – | not run | [#9329](https://github.com/storytold/photocraft/issues/9329) |
 | P82 | Fill a 2 % selection on a full-canvas 24 MP layer + refresh | track only | – | – | – | – | – | not run | [#9329](https://github.com/storytold/photocraft/issues/9329) |
+| P83 | Open a 36 MP JPEG (RGB 8-bit): import (decode + tiles) (#10259) | ≤ 300 ms p95 | – | – | – | – | – | not run | [#10259](https://github.com/storytold/photocraft/issues/10259) |
+| P84 | Open a 36 MP PNG (RGB 8-bit): import (decode + tiles) (#10259) | ≤ 400 ms p95 | – | – | – | – | – | not run | [#10259](https://github.com/storytold/photocraft/issues/10259) |
+| P85 | Open a 36 MP PNG (RGB 16-bit): import (decode + tiles) (#10259) | ≤ 700 ms p95 | – | – | – | – | – | not run | [#10259](https://github.com/storytold/photocraft/issues/10259) |
+| P86 | Open a 36 MP JPEG (RGB 8-bit): import + session admission + first composite (#10259) | ≤ 500 ms p95 | – | – | – | – | – | not run | [#10259](https://github.com/storytold/photocraft/issues/10259) |
 
 **Not measurable yet** (the code can't do it yet; no number is recorded):
 

perf/budgets.toml

@@ -41,6 +41,12 @@ args = ["--cpu", "--group", "layered"]
 quick_args = ["--quick", "--cpu", "--group", "layered"]
 note = "the layered #209 rows on the forced CPU compositor (#9330); run with perf_scenarios so the machine class keeps the GPU"
 
+[[bench]]
+name = "flat_open_bench"
+package = "photocraft-ui-egui"
+quick_args = ["--quick"]
+note = "opening ordinary JPEG/PNG files (24 and 36 MP, 8/16-bit, alpha, EXIF orientation) phase by phase, each case in its own process for its peak memory (#10259)"
+
 [[bench]]
 name = "interactive_bench"
 package = "photocraft-ui-egui"
@@ -790,3 +796,41 @@ issue = 9329
 bench = "perf_scenarios"
 row = "fill a 2 % selection on a full-canvas 24 MP layer + refresh"
 enforce = false  # new; no baseline yet
+
+# ---- #10259: opening ordinary JPEG/PNG files (track only) -------------------------------------
+
+[[scenario]]
+id = "P83"
+title = "Open a 36 MP JPEG (RGB 8-bit): import (decode + tiles) (#10259)"
+issue = 10259
+bench = "flat_open_bench"
+row = "open 36 MP JPEG RGB8: import"
+budget_ms = 300
+enforce = false  # new; no baseline yet
+
+[[scenario]]
+id = "P84"
+title = "Open a 36 MP PNG (RGB 8-bit): import (decode + tiles) (#10259)"
+issue = 10259
+bench = "flat_open_bench"
+row = "open 36 MP PNG RGB8: import"
+budget_ms = 400
+enforce = false  # new; no baseline yet
+
+[[scenario]]
+id = "P85"
+title = "Open a 36 MP PNG (RGB 16-bit): import (decode + tiles) (#10259)"
+issue = 10259
+bench = "flat_open_bench"
+row = "open 36 MP PNG RGB16: import"
+budget_ms = 700
+enforce = false  # new; no baseline yet
+
+[[scenario]]
+id = "P86"
+title = "Open a 36 MP JPEG (RGB 8-bit): import + session admission + first composite (#10259)"
+issue = 10259
+bench = "flat_open_bench"
+row = "open 36 MP JPEG RGB8: editable"
+budget_ms = 500
+enforce = false  # new; no baseline yet