MCP: add export progress and cancellation between photos

Answer ping at photo boundaries, preserve queued requests and EOF completion, and suppress cancelled responses. Keep completed exports and the existing atomic file writer; cover cancellation, output preservation and transport poison recovery.

Co-Authored-By: OpenAI Codex <[email protected]>

apps/lightcraft-cli/src/main.rs

@@ -342,7 +342,7 @@ fn mcp(args: &[String]) -> Result<(), String> {
     eprintln!("lightcraft-cli mcp: serving MCP on stdio ({})", backend.describe());
     let mut server = Server::new(backend).with_command_tools(!compact);
     let stdin = std::io::stdin();
-    server.serve(BufReader::new(stdin.lock()), std::io::stdout().lock()).map_err(|e| e.to_string())
+    server.serve(BufReader::new(stdin), std::io::stdout()).map_err(|e| e.to_string())
 }
 
 fn merge(args: &[String]) -> Result<(), String> {

crates/mcp/src/backend.rs

@@ -15,8 +15,13 @@ pub trait Backend {
     fn has_ui(&self) -> bool;
     /// Short human description ("headless", "remote 127.0.0.1:7980").
     fn describe(&self) -> String;
+    /// Progress/cancellation for the next headless batch export. Other backends ignore it.
+    fn set_progress(&mut self, _hook: Option<ProgressHook>) {}
 }
 
+/// `(completed photos, total photos, next filename)`; false stops before the next photo.
+pub type ProgressHook = Box<dyn FnMut(usize, usize, &str) -> bool + Send>;
+
 /// A running LightCraft app, reached through its loopback control port.
 pub struct Remote {
     addr: String,

crates/mcp/src/headless.rs

@@ -8,7 +8,7 @@ use lightcraft_engine::catalog::PhotoId;
 use lightcraft_raster::Rgba8;
 use serde_json::{Value, json};
 
-use crate::backend::Backend;
+use crate::backend::{Backend, ProgressHook};
 
 /// File extensions recognised as photos when expanding folders.
 pub const PHOTO_EXTENSIONS: &[&str] = &[
@@ -20,6 +20,7 @@ pub const PHOTO_EXTENSIONS: &[&str] = &[
 /// Headless backend: a [`Session`] with filesystem hooks.
 pub struct Headless {
     pub session: Session,
+    progress: Option<ProgressHook>,
 }
 
 impl Drop for Headless {
@@ -37,7 +38,7 @@ impl Default for Headless {
 
 impl Headless {
     pub fn new(session: Session) -> Self {
-        Self { session }
+        Self { session, progress: None }
     }
 
     /// A headless session with the procedurally generated demo library.
@@ -60,7 +61,7 @@ impl Headless {
     /// (see `lightcraft_engine::export::ExportOptions::from_json`, plus `ids`, `dir`, `path`).
     /// With `path` and no `format`, the format follows the path's extension.
     fn export(&mut self, p: &Value) -> Result<Value, String> {
-        use lightcraft_engine::export::{Destination, ExportFormat, ExportOptions, Resize, export_batch};
+        use lightcraft_engine::export::{Destination, ExportFormat, ExportOptions, Resize, prepare_batch, run_batch};
         let p = &self.session.export_params(p)?;
         let mut opts = ExportOptions::from_params(p).map_err(|e| e.to_string())?;
         if !ExportOptions::has_size_param(p) {
@@ -82,10 +83,28 @@ impl Headless {
         };
         let dir = p.get("dir").and_then(Value::as_str).unwrap_or("");
         let write = &mut lightcraft_engine::export::write_file;
-        let files =
-            export_batch(&mut self.session, &ids, &opts, &Destination { dir: dir.to_string(), exact: exact.map(str::to_string) }, write, &|path| {
-                Path::new(path).exists()
-            })?;
+        let items = prepare_batch(&mut self.session, &ids, &opts)?;
+        let total = items.len();
+        let mut hook = self.progress.take();
+        let mut cancelled = false;
+        let files = run_batch(
+            items,
+            &opts,
+            &Destination { dir: dir.to_string(), exact: exact.map(str::to_string) },
+            write,
+            &|path| Path::new(path).exists(),
+            true,
+            &mut |done, name| {
+                cancelled = hook.as_mut().is_some_and(|h| !h(done, total, name));
+                !cancelled
+            },
+        )?;
+        if !cancelled && let Some(hook) = hook.as_mut() {
+            cancelled = !hook(total, total, "");
+        }
+        if cancelled {
+            return Err("cancelled".into());
+        }
         // Single-photo exports also report path/width/height at the top level (back-compat).
         let mut out = files.first().cloned().unwrap_or_else(|| json!({}));
         out["files"] = json!(files);
@@ -94,6 +113,10 @@ impl Headless {
 }
 
 impl Backend for Headless {
+    fn set_progress(&mut self, hook: Option<ProgressHook>) {
+        self.progress = hook;
+    }
+
     fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
         let p = if params.is_null() { json!({}) } else { params };
         match method {

crates/mcp/src/lib.rs

@@ -21,7 +21,7 @@ mod headless;
 mod server;
 mod tools;
 
-pub use backend::{Backend, Remote};
+pub use backend::{Backend, ProgressHook, Remote};
 pub use base64::{base64_decode, base64_encode};
 pub use headless::{Headless, PHOTO_EXTENSIONS, encode_image, expand_paths, write_image};
 pub use server::{PROTOCOL_VERSION, Server};

crates/mcp/src/progress.rs

@@ -0,0 +1,165 @@
+//! Stdio progress and cancellation between photos. The backend keeps its session on the
+//! serving thread; only stdin and stdout move to transport threads. Other calls wait in order.
+
+use std::collections::VecDeque;
+use std::io::{BufRead, Write};
+use std::sync::atomic::Ordering;
+use std::sync::mpsc::{Receiver, Sender, TryRecvError};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use serde_json::{Value, json};
+
+use super::{Server, response};
+use crate::backend::ProgressHook;
+
+type Input = Option<std::io::Result<String>>;
+
+#[derive(Clone)]
+pub(super) struct Wire {
+    inbox: Arc<Mutex<Receiver<Input>>>,
+    out: Sender<String>,
+    deferred: Arc<Mutex<VecDeque<Input>>>,
+}
+
+fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
+    m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+impl Server {
+    pub(super) fn serve_wire(&mut self, input: impl BufRead + Send, output: impl Write + Send) -> std::io::Result<()> {
+        std::thread::scope(|scope| {
+            let (in_tx, in_rx) = std::sync::mpsc::channel::<Input>();
+            let (out_tx, out_rx) = std::sync::mpsc::channel::<String>();
+            let wire = Wire { inbox: Arc::new(Mutex::new(in_rx)), out: out_tx, deferred: Arc::new(Mutex::new(VecDeque::new())) };
+            let writer = std::thread::Builder::new().name("mcp-output".into()).spawn_scoped(scope, move || -> std::io::Result<()> {
+                let mut output = output;
+                for line in out_rx {
+                    output.write_all(line.as_bytes())?;
+                    output.write_all(b"\n")?;
+                    output.flush()?;
+                }
+                Ok(())
+            })?;
+            let reader = match std::thread::Builder::new().name("mcp-input".into()).spawn_scoped(scope, move || {
+                for line in input.lines() {
+                    let failed = line.is_err();
+                    if in_tx.send(Some(line)).is_err() || failed {
+                        return;
+                    }
+                }
+                let _ = in_tx.send(None);
+            }) {
+                Ok(reader) => reader,
+                Err(error) => {
+                    drop(wire);
+                    let _ = writer.join();
+                    return Err(error);
+                }
+            };
+            self.wire = Some(wire.clone());
+            let served = (|| -> std::io::Result<()> {
+                loop {
+                    let queued = lock(&wire.deferred).pop_front();
+                    let line = match queued {
+                        Some(line) => line,
+                        None => lock(&wire.inbox).recv().ok().flatten(),
+                    };
+                    let Some(line) = line else { return Ok(()) };
+                    if let Some(reply) = self.handle_line(&line?) {
+                        wire.out.send(reply).map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "MCP output closed"))?;
+                    }
+                }
+            })();
+            self.wire = None;
+            drop(wire);
+            let written = writer.join().map_err(|_| std::io::Error::other("MCP output thread panicked")).and_then(|r| r);
+            let read = reader.join().map_err(|_| std::io::Error::other("MCP input thread panicked"));
+            served.and(written).and(read)
+        })
+    }
+
+    /// Called at photo boundaries, including completion. EOF finishes the pending export;
+    /// cancellation stops before the next photo. No file scan or output deletion is needed:
+    /// the existing writer replaces each whole file atomically.
+    pub(super) fn progress_hook(&self) -> Option<ProgressHook> {
+        let wire = self.wire.clone()?;
+        let (id, token) = self.current.clone()?;
+        let suppress = self.suppress.clone();
+        let mut closed = false;
+        let mut last = None;
+        let mut reported: Option<Instant> = None;
+        Some(Box::new(move |done, total, name| {
+            while !closed {
+                let next = lock(&wire.inbox).try_recv();
+                let line = match next {
+                    Ok(Some(Ok(line))) => line,
+                    Ok(Some(Err(error))) => {
+                        lock(&wire.deferred).push_back(Some(Err(error)));
+                        closed = true;
+                        break;
+                    }
+                    Ok(None) | Err(TryRecvError::Disconnected) => {
+                        lock(&wire.deferred).push_back(None);
+                        closed = true;
+                        break;
+                    }
+                    Err(TryRecvError::Empty) => break,
+                };
+                let msg: Value = serde_json::from_str(&line).unwrap_or(Value::Null);
+                match msg.get("method").and_then(Value::as_str) {
+                    Some("notifications/cancelled") if msg.get("id").is_none() && msg["params"]["requestId"] == id => {
+                        suppress.store(true, Ordering::Relaxed);
+                    }
+                    Some("ping")
+                        if msg.get("id").is_some_and(|i| i.is_string() || i.is_number())
+                            && msg.get("params").is_none_or(|p| p.is_null() || p.is_object()) =>
+                    {
+                        let _ = wire.out.send(response(msg["id"].clone(), json!({})).to_string());
+                    }
+                    _ => lock(&wire.deferred).push_back(Some(Ok(line))),
+                }
+            }
+            if suppress.load(Ordering::Relaxed) {
+                return false;
+            }
+            if let Some(token) = &token
+                && total > 0
+                && last.is_none_or(|last| done > last)
+                && (done == total || reported.is_none_or(|at| at.elapsed() >= Duration::from_millis(100)))
+            {
+                last = Some(done);
+                reported = Some(Instant::now());
+                let message = if done < total {
+                    format!("Exporting {name} ({} of {total})", done.saturating_add(1))
+                } else {
+                    format!("Exported {total} photo(s)")
+                };
+                let notification = json!({"jsonrpc":"2.0","method":"notifications/progress",
+                    "params":{"progressToken":token,"progress":done,"total":total,"message":message}});
+                let _ = wire.out.send(notification.to_string());
+            }
+            true
+        }))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn transport_lock_recovers_after_panic() {
+        let state = Arc::new(Mutex::new(VecDeque::new()));
+        let held = state.clone();
+        let _ = std::thread::spawn(move || {
+            held.lock().unwrap().push_back(7);
+            let _guard = held.lock().unwrap();
+            panic!("synthetic transport panic");
+        })
+        .join();
+        assert_eq!(lock(&state).pop_front(), Some(7));
+        lock(&state).push_back(8);
+        assert_eq!(lock(&state).pop_front(), Some(8));
+    }
+}

crates/mcp/src/server.rs

@@ -1,6 +1,11 @@
 //! JSON-RPC 2.0 framing and the MCP lifecycle / tools / resources methods.
 
 use std::io::{BufRead, Write};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+#[path = "progress.rs"]
+mod progress;
 
 use serde_json::{Value, json};
 
@@ -40,6 +45,9 @@ pub struct Server {
     backend: Box<dyn Backend>,
     initialized: bool,
     command_tools: bool,
+    wire: Option<progress::Wire>,
+    current: Option<(Value, Option<Value>)>,
+    suppress: Arc<AtomicBool>,
 }
 
 fn response(id: Value, result: Value) -> Value {
@@ -56,7 +64,7 @@ fn resource(uri: &str, name: &str, title: &str, description: &str) -> Value {
 
 impl Server {
     pub fn new(backend: Box<dyn Backend>) -> Self {
-        Self { backend, initialized: false, command_tools: true }
+        Self { backend, initialized: false, command_tools: true, wire: None, current: None, suppress: Arc::new(AtomicBool::new(false)) }
     }
 
     /// Whether `tools/list` includes one `cmd_*` tool per registered command (default true).
@@ -77,16 +85,8 @@ impl Server {
 
     /// Serve newline-delimited JSON-RPC until `input` closes. Logs go to stderr only (stdout is
     /// the protocol stream).
-    pub fn serve(&mut self, input: impl BufRead, mut output: impl Write) -> std::io::Result<()> {
-        for line in input.lines() {
-            let line = line?;
-            if let Some(reply) = self.handle_line(&line) {
-                output.write_all(reply.as_bytes())?;
-                output.write_all(b"\n")?;
-                output.flush()?;
-            }
-        }
-        Ok(())
+    pub fn serve(&mut self, input: impl BufRead + Send, output: impl Write + Send) -> std::io::Result<()> {
+        self.serve_wire(input, output)
     }
 
     /// Handle one line; returns the reply line (None for notifications and blank lines).
@@ -133,7 +133,15 @@ impl Server {
         if !(params.is_object() || params.is_null()) {
             return Some(error(id, INVALID_PARAMS, "`params` must be an object"));
         }
-        Some(match self.request(method, &params) {
+        let token = params.get("_meta").and_then(|m| m.get("progressToken")).filter(|t| t.is_string() || t.is_number()).cloned();
+        self.current = Some((id.clone(), token));
+        self.suppress.store(false, Ordering::Relaxed);
+        let result = self.request(method, &params);
+        self.current = None;
+        if self.suppress.swap(false, Ordering::Relaxed) {
+            return None;
+        }
+        Some(match result {
             Ok(mut r) => {
                 modern_result_fields(method, &params, &mut r);
                 response(id, r)
@@ -172,7 +180,20 @@ impl Server {
                 let name = params.get("name").and_then(Value::as_str).ok_or((INVALID_PARAMS, "missing tool `name`".to_string()))?;
                 let args = params.get("arguments").cloned().unwrap_or(Value::Null);
                 check_args(name, &args).map_err(|e| (INVALID_PARAMS, e))?;
-                Ok(self.guarded(|b| call_tool(b, name, &args)).unwrap_or_else(ToolResult::error).to_value())
+                let long = matches!(name, "export" | "cmd_app_export")
+                    || (name == "command_run" && args["id"] == "app.export")
+                    || (name == "run_command" && args["command"] == "app.export");
+                let hook = if long && !self.backend.has_ui() { self.progress_hook() } else { None };
+                let result = self.guarded(|b| {
+                    b.set_progress(hook);
+                    call_tool(b, name, &args)
+                });
+                let clear = self.guarded(|b| b.set_progress(None));
+                let result = match (result, clear) {
+                    (Ok(result), Ok(())) => result,
+                    (Err(error), _) | (_, Err(error)) => ToolResult::error(error),
+                };
+                Ok(result.to_value())
             }
             "resources/list" => Ok(json!({"resources": [
                 resource(DOCUMENT_URI, "document", "Library document", "Library state and counts (same as doc_inspect)"),

crates/mcp/tests/e2e.rs

@@ -153,3 +153,115 @@ fn remote_unreachable_is_a_tool_error() {
     assert_eq!(r[0]["result"]["isError"], true);
     assert!(r[0]["result"]["content"][0]["text"].as_str().unwrap().contains("not reachable"));
 }
+
+/// Real stdio framing and real atomic exports: progress, ping during a batch, cancellation
+/// between complete photos, unrelated output preservation, and a usable session afterwards.
+#[test]
+fn export_progress_and_cancel() {
+    let dir = std::env::temp_dir().join(format!("lc-mcp-progress-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    let (input, mut send) = std::io::pipe().unwrap();
+    let (output, write) = std::io::pipe().unwrap();
+    let server = std::thread::spawn(move || Server::new(Box::new(Headless::demo())).serve(BufReader::new(input), write).unwrap());
+    let (tx, rx) = std::sync::mpsc::channel();
+    let reader = std::thread::spawn(move || {
+        for line in BufReader::new(output).lines() {
+            let value: Value = serde_json::from_str(&line.unwrap()).unwrap();
+            if tx.send(value).is_err() {
+                break;
+            }
+        }
+    });
+    let next = || rx.recv_timeout(std::time::Duration::from_secs(30)).expect("MCP reply/progress");
+    writeln!(send, "{}", call(1, "query_photos", json!({"limit":8}))).unwrap();
+    let query = next();
+    let ids: Vec<_> = query["result"]["structuredContent"]["photos"].as_array().unwrap().iter().map(|p| p["id"].clone()).collect();
+    assert_eq!(ids.len(), 8);
+    let mut request = call(2, "export", json!({"ids":ids,"dir":dir.join("complete"),"longEdge":1200,"format":"png"}));
+    request["params"]["_meta"] = json!({"progressToken":"photos","io.modelcontextprotocol/protocolVersion":"2026-07-28"});
+    writeln!(send, "{request}").unwrap();
+    let (mut last, mut notes, mut asked, mut answered) = (-1.0, 0, false, false);
+    loop {
+        let value = next();
+        if value["method"] == "notifications/progress" {
+            assert_eq!(value["params"]["progressToken"], "photos");
+            let progress = value["params"]["progress"].as_f64().unwrap();
+            assert!(progress > last && progress <= 8.0, "{value}");
+            assert_eq!(value["params"]["total"], 8);
+            last = progress;
+            notes += 1;
+            if !asked {
+                writeln!(send, "{}", json!({"jsonrpc":"2.0","id":3,"method":"ping"})).unwrap();
+                asked = true;
+            }
+        } else if value["id"] == 3 {
+            answered = true;
+        } else {
+            assert_eq!(value["id"], 2, "{value}");
+            assert_eq!(value["result"]["isError"], false, "{value}");
+            assert_eq!(value["result"]["resultType"], "complete");
+            break;
+        }
+    }
+    assert!(notes >= 2 && answered, "{notes} progress messages; ping answered {answered}");
+    assert_eq!(last, 8.0);
+    let out = dir.join("cancelled");
+    std::fs::create_dir_all(&out).unwrap();
+    let decoy = out.join("photo999.png");
+    std::fs::write(&decoy, b"unrelated existing output").unwrap();
+    let mut request = call(4, "command_run", json!({"id":"app.export","params":{"ids":ids,"dir":out,"longEdge":1200,"format":"png"}}));
+    request["params"]["_meta"] = json!({"progressToken":44});
+    writeln!(send, "{request}").unwrap();
+    loop {
+        let value = next();
+        assert_ne!(value["id"], 4, "must be cancellable before completion: {value}");
+        assert_eq!(value["params"]["progressToken"], 44, "{value}");
+        if value["params"]["progress"].as_f64().unwrap() > 0.0 {
+            break;
+        }
+    }
+    writeln!(send, "{}", json!({"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":4}})).unwrap();
+    writeln!(send, "{}", call(5, "doc_inspect", json!({}))).unwrap();
+    loop {
+        let value = next();
+        if value["method"] == "notifications/progress" {
+            continue;
+        }
+        assert_eq!(value["id"], 5, "cancelled request must not reply: {value}");
+        assert_eq!(value["result"]["isError"], false);
+        break;
+    }
+    assert_eq!(std::fs::read(&decoy).unwrap(), b"unrelated existing output");
+    let photos: Vec<_> = std::fs::read_dir(&out).unwrap().map(|e| e.unwrap().path()).filter(|p| *p != decoy).collect();
+    assert!(!photos.is_empty() && photos.len() < ids.len(), "only completed photos remain: {photos:?}");
+    for path in &photos {
+        lightcraft_codecs::decode(&std::fs::read(path).unwrap(), Default::default()).expect("complete image, no partial/temp file");
+    }
+    assert_eq!(std::fs::read_dir(dir.join("complete")).unwrap().count(), 8, "earlier export retained");
+    drop(send);
+    server.join().unwrap();
+    reader.join().unwrap();
+    std::fs::remove_dir_all(dir).unwrap();
+}
+
+#[test]
+fn export_without_token_finishes_after_eof() {
+    let dir = std::env::temp_dir().join(format!("lc-mcp-eof-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    let mut headless = Headless::demo();
+    let ids: Vec<_> = headless.session.visible_cloned().iter().take(2).map(|id| id.0).collect();
+    let request = call(1, "cmd_app_export", json!({"ids":ids,"dir":dir,"longEdge":64,"format":"png"}));
+    // Headless sessions remain movable to a worker after installing progress support.
+    let replies = std::thread::spawn(move || {
+        let mut s = Server::new(Box::new(headless));
+        session(&mut s, &[request, call(2, "doc_inspect", json!({}))])
+    })
+    .join()
+    .unwrap();
+    assert_eq!(replies.len(), 2, "no token means no notifications");
+    assert_eq!(replies[0]["result"]["isError"], false, "{replies:?}");
+    assert_eq!(replies[1]["id"], 2, "queued request survives EOF");
+    assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 2);
+    std::fs::remove_dir_all(dir).unwrap();
+}

docs/mcp.md

@@ -219,3 +219,20 @@ lightcraft-cli calibrate --max 300 ~/Pictures/2026   # camera colour profiles (d
   (`Remote`) against a stand-in control server; also import → render → JPEG export of a real file.
 - `apps/lightcraft-cli/tests/cli.rs` — spawns `lightcraft-cli mcp` with real pipes; `render`;
   `commands`.
+
+## Export progress and cancellation
+
+Headless direct `export`, `command_run` / `run_command` with `app.export`, and `cmd_app_export`
+calls report photo-count progress when `params._meta.progressToken` is a string or number.
+Notifications are strictly increasing, at most ten per second plus the final total. No token
+means no notifications. `ping` is answered at photo boundaries; other requests wait in order
+until the export ends. EOF lets a pending export finish and preserves queued requests.
+
+`notifications/cancelled` with `params.requestId` stops the matching export before its next
+photo, suppressing its response. Unknown/completed request ids are ignored. A photo already
+being processed finishes first: completed photos remain, each written with the existing atomic
+file writer, and there are no partial files to delete. Unrelated outputs and earlier exports
+are untouched. A failed export is an `isError` tool result. Transport mutexes recover poisoning.
+
+Connect mode and exports inside `command_batch` remain synchronous and do not report MCP
+progress or cancellation. Use a direct headless export call for this behavior.