diff --git a/apps/lightcraft-cli/src/main.rs b/apps/lightcraft-cli/src/main.rs index f8ddf550d..7c9d255e9 100644 --- a/apps/lightcraft-cli/src/main.rs +++ b/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> { diff --git a/crates/mcp/src/backend.rs b/crates/mcp/src/backend.rs index 5db2aaaa7..4772363b7 100644 --- a/crates/mcp/src/backend.rs +++ b/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) {} } +/// `(completed photos, total photos, next filename)`; false stops before the next photo. +pub type ProgressHook = Box bool + Send>; + /// A running LightCraft app, reached through its loopback control port. pub struct Remote { addr: String, diff --git a/crates/mcp/src/headless.rs b/crates/mcp/src/headless.rs index 837302d60..0817b39de 100644 --- a/crates/mcp/src/headless.rs +++ b/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, } 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 { - 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) { + self.progress = hook; + } + fn call(&mut self, method: &str, params: Value) -> Result { let p = if params.is_null() { json!({}) } else { params }; match method { diff --git a/crates/mcp/src/lib.rs b/crates/mcp/src/lib.rs index a1fb5510d..b2417fe03 100644 --- a/crates/mcp/src/lib.rs +++ b/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}; diff --git a/crates/mcp/src/progress.rs b/crates/mcp/src/progress.rs new file mode 100644 index 000000000..2043dfd91 --- /dev/null +++ b/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>; + +#[derive(Clone)] +pub(super) struct Wire { + inbox: Arc>>, + out: Sender, + deferred: Arc>>, +} + +fn lock(m: &Mutex) -> 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::(); + let (out_tx, out_rx) = std::sync::mpsc::channel::(); + 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 { + 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 = 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)); + } +} diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs index 3f1f12f0c..ba64ef45a 100644 --- a/crates/mcp/src/server.rs +++ b/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, initialized: bool, command_tools: bool, + wire: Option, + current: Option<(Value, Option)>, + suppress: Arc, } 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) -> 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, ¶ms) { + 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, ¶ms); + self.current = None; + if self.suppress.swap(false, Ordering::Relaxed) { + return None; + } + Some(match result { Ok(mut r) => { modern_result_fields(method, ¶ms, &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)"), diff --git a/crates/mcp/tests/e2e.rs b/crates/mcp/tests/e2e.rs index ea27919c0..ef2e49b6b 100644 --- a/crates/mcp/tests/e2e.rs +++ b/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(); +} diff --git a/docs/mcp.md b/docs/mcp.md index 24f53c659..3134468e3 100644 --- a/docs/mcp.md +++ b/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.