MCP: cancel long exports, answer while one runs, report progress (#2165, #2171, #2172)

The MCP server read one line, ran it and answered it before reading the next,
so notifications/cancelled was never seen during a call and every request
(even ping) waited behind a running export.

- vectorcraft_doc::cancel: a thread-local cancel token and progress sink that
  long work polls. Checkpoints: PDF pages, PDF flattening, geometry bake,
  raster-effect flattening, drawing objects, raster artboards, and before any
  file is written. A cancelled command returns EngineError::Cancelled and
  changes nothing.
- Server::serve reads on its own thread: notifications/cancelled drops a
  queued request or stops the running one (no reply, as the MCP spec says),
  engine-free requests (ping, the list methods, prompts/get) are answered at
  once, a request that would wait behind a call running for 2 s is refused
  with -32003 naming that call, and _meta.progressToken gets
  notifications/progress (page n of m). Queue and output are bounded.
- The remote backend stops waiting for the app when the call is cancelled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/vectorcraft-cli/src/main.rs

@@ -213,8 +213,8 @@ fn mcp(args: &[String]) -> Result<(), String> {
     // uses `vectorcraft_mcp` free to bring its own. Silent until a client sends logging/setLevel.
     vectorcraft_mcp::logging::install();
     let stdin = std::io::stdin();
-    let stdout = std::io::stdout();
-    Server::new(backend).serve(stdin.lock(), stdout.lock()).map_err(|e| e.to_string())
+    // The server reads and writes from threads of its own, so it takes the handles, not locks held here.
+    Server::new(backend).serve(std::io::BufReader::new(stdin), std::io::stdout()).map_err(|e| e.to_string())
 }
 
 fn commands() -> Result<(), String> {

crates/doc/src/cancel.rs

@@ -0,0 +1,142 @@
+//! Cooperative cancellation and progress for long-running work (exports, flattening, baking).
+//!
+//! A host that wants to be able to stop a call (the MCP server, on `notifications/cancelled`)
+//! runs it inside [`scope`] with a [`CancelToken`]. Code deep below, which has no handle on the
+//! host, calls [`check`] (or [`is_cancelled`] where it can't return an error) at points where
+//! stopping is safe, and returns [`Cancelled`] up through the same `Result`s it already has. The
+//! token is tied to the thread that entered the scope, so concurrent sessions on other threads
+//! (and tests running in parallel) never see each other's cancellation, and code outside any
+//! scope is never cancelled.
+//!
+//! Work that stops early leaves nothing behind: callers drop whatever was half built and report
+//! the error, so a cancelled command changes nothing.
+
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+/// A flag one thread raises and another polls. Cloning shares the flag.
+#[derive(Clone, Debug, Default)]
+pub struct CancelToken(Arc<AtomicBool>);
+
+impl CancelToken {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Ask for the work holding this token to stop.
+    pub fn cancel(&self) {
+        self.0.store(true, Ordering::Relaxed);
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+/// The error of work that stopped because it was cancelled.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Cancelled;
+
+impl std::fmt::Display for Cancelled {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("cancelled")
+    }
+}
+
+impl std::error::Error for Cancelled {}
+
+/// Receives progress: `(done, total, what is being done)`. `total` is 0 when it isn't known.
+pub type ProgressSink = Rc<dyn Fn(u64, u64, &str)>;
+
+struct Scope {
+    token: CancelToken,
+    sink: Option<ProgressSink>,
+}
+
+thread_local! {
+    static CURRENT: RefCell<Option<Scope>> = const { RefCell::new(None) };
+}
+
+/// Puts the previous scope back, also when the work inside unwinds.
+struct Restore(Option<Scope>);
+
+impl Drop for Restore {
+    fn drop(&mut self) {
+        let previous = self.0.take();
+        // `try_with`: during thread teardown the slot may be gone; nothing to restore then.
+        let _ = CURRENT.try_with(|c| *c.borrow_mut() = previous);
+    }
+}
+
+/// Run `f` with `token` (and an optional progress `sink`) in effect on this thread.
+pub fn scope<T>(token: &CancelToken, sink: Option<ProgressSink>, f: impl FnOnce() -> T) -> T {
+    let previous = CURRENT.with(|c| c.borrow_mut().replace(Scope { token: token.clone(), sink }));
+    let _restore = Restore(previous);
+    f()
+}
+
+/// Has the work running on this thread been cancelled? Always false outside a [`scope`].
+pub fn is_cancelled() -> bool {
+    CURRENT.with(|c| c.borrow().as_ref().is_some_and(|s| s.token.is_cancelled()))
+}
+
+/// Is this thread running inside a [`scope`], so that a cancel can reach it?
+pub fn is_cancellable() -> bool {
+    CURRENT.with(|c| c.borrow().is_some())
+}
+
+/// `Err(Cancelled)` when the work running on this thread has been cancelled.
+pub fn check() -> Result<(), Cancelled> {
+    if is_cancelled() { Err(Cancelled) } else { Ok(()) }
+}
+
+/// Report that `done` of `total` steps (0: unknown) are finished. A no-op outside a [`scope`]
+/// that has a sink.
+pub fn progress(done: u64, total: u64, what: &str) {
+    // The sink runs outside the borrow, so it may itself call back into this module.
+    let sink = CURRENT.with(|c| c.borrow().as_ref().and_then(|s| s.sink.clone()));
+    if let Some(sink) = sink {
+        sink(done, total, what);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn only_work_inside_the_scope_is_cancelled() {
+        let t = CancelToken::new();
+        assert_eq!(check(), Ok(()));
+        scope(&t, None, || {
+            assert_eq!(check(), Ok(()));
+            t.cancel();
+            assert_eq!(check(), Err(Cancelled));
+            // A nested scope with its own token shadows, then restores.
+            scope(&CancelToken::new(), None, || assert_eq!(check(), Ok(())));
+            assert!(is_cancelled());
+        });
+        assert!(!is_cancelled(), "the scope ended");
+    }
+
+    #[test]
+    fn progress_reaches_the_sink_and_the_scope_survives_a_panic() {
+        let seen = Rc::new(RefCell::new(vec![]));
+        let sink: ProgressSink = {
+            let seen = seen.clone();
+            Rc::new(move |d, t, w| seen.borrow_mut().push((d, t, w.to_string())))
+        };
+        let t = CancelToken::new();
+        let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            scope(&t, Some(sink), || {
+                progress(1, 3, "page");
+                panic!("boom");
+            })
+        }));
+        assert!(r.is_err());
+        progress(2, 3, "after");
+        assert_eq!(*seen.borrow(), vec![(1, 3, "page".to_string())]);
+    }
+}

crates/doc/src/lib.rs

@@ -8,6 +8,7 @@
 pub mod appearance;
 pub mod assets;
 pub mod blend;
+pub mod cancel;
 pub mod clipnest;
 pub mod cmyk;
 pub mod corners;

crates/effects/src/bake.rs

@@ -89,7 +89,8 @@ fn path_kind(d: &mut Document, path: PathData, rule: FillRule) -> NodeKind {
 
 /// `n` with its geometry effects evaluated (`None` = unchanged).
 fn bake_node(d: &mut Document, n: &Node) -> Option<Node> {
-    if !needs_bake(n) {
+    // A cancelled export leaves the rest of the document as it is: the caller drops the result.
+    if !needs_bake(n) || vectorcraft_doc::cancel::is_cancelled() {
         return None;
     }
     // Colour adjustments: the object recoloured (inside it too), then baked as it is.

crates/engine/src/cmd/fileio/encode.rs

@@ -413,6 +413,8 @@ fn encode_all_of(doc: &Document, format: &str, p: &Value) -> Result<Encoded> {
     let f = super::writable(C, Some(format), None)?;
     let failed = vectorcraft_render::decode_failures();
     let mut enc = encode_files(doc, f, p)?;
+    // Baking, flattening and drawing stop early when cancelled and leave a truncated result.
+    vectorcraft_doc::cancel::check()?;
     // An image the renderer couldn't decode draws nothing: say so rather than write a picture
     // that silently lacks it.
     if f.raster && vectorcraft_render::decode_failures() > failed {
@@ -502,7 +504,10 @@ fn encode_files(doc: &Document, f: &Format, p: &Value) -> Result<Encoded> {
             };
             let mut enc = Encoded::default();
             let mut renderer = vectorcraft_render::Renderer::new();
-            for b in chosen {
+            let total = chosen.len() as u64;
+            for (done, b) in chosen.into_iter().enumerate() {
+                vectorcraft_doc::cancel::check()?;
+                vectorcraft_doc::cancel::progress(done as u64, total, &format!("Rendering artboard {} of {total}", done + 1));
                 let region = doc.artboards.get(b).ok_or_else(|| bad(C, format!("no artboard {}", b + 1)))?.rect;
                 check_format_size(f, region.width() * scale, region.height() * scale)?;
                 vectorcraft_render::raster_size(region, scale).map_err(|e| bad(C, e))?;

crates/engine/src/cmd/fileio/mod.rs

@@ -861,6 +861,8 @@ pub(crate) fn read_file_head(path: &str, max: usize) -> Result<Vec<u8>> {
 
 #[cfg(not(target_arch = "wasm32"))]
 pub(crate) fn write_file(path: &str, bytes: &[u8]) -> Result<()> {
+    // Work that stops early for a cancel can leave a truncated result: none of it is written.
+    vectorcraft_doc::cancel::check()?;
     may_write(path)?;
     // Never a half-written file: see [`write_atomic`].
     write_atomic(std::path::Path::new(path), bytes).map_err(|e| EngineError::Other(format!("{path}: {e}")))
@@ -869,6 +871,7 @@ pub(crate) fn write_file(path: &str, bytes: &[u8]) -> Result<()> {
 /// [`write_file`] with what `fill` writes into the file (a save streamed into it).
 #[cfg(not(target_arch = "wasm32"))]
 pub fn write_file_with(path: &str, fill: &mut dyn FnMut(&mut dyn std::io::Write) -> std::io::Result<()>) -> Result<()> {
+    vectorcraft_doc::cancel::check()?;
     may_write(path)?;
     vectorcraft_format::write_atomic_with(std::path::Path::new(path), |f| fill(f)).map_err(|e| EngineError::Other(format!("{path}: {e}")))
 }

crates/engine/src/cmd/fileio/pdf.rs

@@ -169,6 +169,7 @@ pub(crate) fn merge(base: &mut Value, over: &Value) {
 pub(crate) fn pdf_error(cmd: &str, e: PdfError) -> EngineError {
     match e {
         PdfError::BadSetting(_) | PdfError::Unsupported(_) => bad(cmd, e.to_string()),
+        PdfError::Cancelled => EngineError::Cancelled,
         e => EngineError::Other(e.to_string()),
     }
 }
@@ -273,7 +274,11 @@ pub(super) fn encode_carrying(cmd: &str, doc: &Document, p: &Value, native: impl
         }
     }
     let set = &opts.settings;
+    if set.pdf13() {
+        vectorcraft_doc::cancel::progress(0, 0, "Flattening transparency");
+    }
     let flat = flatten_for(cmd, doc, set, p)?;
+    vectorcraft_doc::cancel::check()?;
     if flat.is_some() {
         let kind = if set.standard.flattens() { set.standard.label() } else { set.compatibility.label() };
         warnings.push(format!("transparency is flattened into opaque art and images ({kind} files have none): see the flattener preset"));

crates/engine/src/cmd/flatten.rs

@@ -818,7 +818,10 @@ fn plan(src: &Document, roots: &[NodeId], o: &FlattenOptions, render: bool) -> R
     let see_through: Vec<bool> = plain.iter().map(|n| n.shows_transparency() || o.draws_see_through_image(n)).collect();
     let cmyk = src.color_mode == ColorMode::Cmyk;
     let mut out = Plan { flat: vec![], kept: vec![], transparent: roots.iter().zip(&see_through).filter(|(_, t)| **t).map(|(id, _)| *id).collect() };
+    // Baking above stopped early when cancelled; each group is a place to stop (a plan changes nothing).
+    vectorcraft_doc::cancel::check()?;
     for g in overlapping(&reaches) {
+        vectorcraft_doc::cancel::check()?;
         let ids: Vec<NodeId> = g.iter().map(|&i| roots[i]).collect();
         let group = g.iter().any(|&i| see_through[i]).then(|| {
             let plain: Vec<&Node> = g.iter().map(|&i| &plain[i]).collect();
@@ -950,6 +953,8 @@ fn render_raster(doc: &Document, art: &[Node], rect: Rect, ppi: f64, clip: Optio
     );
     let mut r = vectorcraft_render::Renderer::new();
     let mut img = r.render_region(&isolated_doc(doc, art.to_vec()), region, scale, !o.preserve_alpha);
+    // The drawing stops early when cancelled: this image is not whole.
+    vectorcraft_doc::cancel::check()?;
     if o.no_transparency.is_some() && clip.is_some() && !o.preserve_alpha {
         // Without transparency the clip alone shapes the image: it is opaque, over white (its
         // edge pixels too).

crates/engine/src/cmd/rasterfx.rs

@@ -440,6 +440,10 @@ fn walk_mask(out: &mut Document, n: &mut Node, report: &mut FlattenReport) {
 /// has none. An object whose effects all paint below it keeps its vector art over an image of
 /// them; one whose effects change it (blur, feather, inner glow) becomes the image.
 fn walk(out: &mut Document, n: &Node, report: &mut FlattenReport) -> Option<Node> {
+    // A cancelled export leaves the rest of the document as it is: the caller drops the result.
+    if vectorcraft_doc::cancel::is_cancelled() {
+        return None;
+    }
     let checkpoint = report.checkpoint();
     let result = walk_inner(out, n, report);
     report.attribute_since(checkpoint, n);
@@ -558,6 +562,8 @@ pub fn export_pdf_with_report(
 ) -> std::result::Result<vectorcraft_pdf::ExportReport, vectorcraft_pdf::PdfError> {
     let mut report = FlattenReport::new(Which::All, doc);
     let flat = flatten(doc, &mut report);
+    // Flattening stops early when cancelled, leaving a half-flattened copy that must not be written.
+    vectorcraft_doc::cancel::check().map_err(|_| vectorcraft_pdf::PdfError::Cancelled)?;
     let mut exported = vectorcraft_pdf::export_with_report(flat.as_ref().unwrap_or(doc), opts)?;
     exported.warnings.extend(report.warnings());
     Ok(exported)

crates/engine/src/lib.rs

@@ -34,6 +34,7 @@ pub use cmd::{CommandInfo, CommandSpec, command_specs, find_command};
 pub use journal::Journal;
 pub use tooling::{UiRequest, ViewInfo};
 pub use vectorcraft_doc as doc;
+pub use vectorcraft_doc::cancel;
 pub use vectorcraft_tools as tools;
 
 #[derive(Debug, thiserror::Error)]
@@ -50,11 +51,20 @@ pub enum EngineError {
     NoNode(NodeId),
     #[error("{0}")]
     Other(String),
+    /// The caller asked for the command to stop ([`cancel`]). It changed nothing.
+    #[error("cancelled; nothing was changed")]
+    Cancelled,
     /// A bug: the command panicked. The document was rolled back to its state before the command.
     #[error("internal error in `{cmd}`: {msg} (the document was left as it was before; please report this bug)")]
     Internal { cmd: String, msg: String },
 }
 
+impl From<cancel::Cancelled> for EngineError {
+    fn from(_: cancel::Cancelled) -> Self {
+        EngineError::Cancelled
+    }
+}
+
 impl From<vectorcraft_doc::DocError> for EngineError {
     fn from(e: vectorcraft_doc::DocError) -> Self {
         EngineError::Other(e.to_string())
@@ -1892,6 +1902,8 @@ mod tests_brushsym;
 #[cfg(test)]
 mod tests_build;
 #[cfg(test)]
+mod tests_cancel;
+#[cfg(test)]
 mod tests_charstroke;
 #[cfg(test)]
 mod tests_clip;

crates/engine/src/tests_cancel.rs

@@ -0,0 +1,90 @@
+//! Cancelling a long export (#2165): it stops at the next checkpoint with `EngineError::Cancelled`,
+//! writes nothing, leaves the document as it was, and the session goes on working.
+
+use std::cell::Cell;
+use std::rc::Rc;
+
+use serde_json::json;
+
+use crate::cancel::{self, CancelToken, ProgressSink};
+use crate::{EngineError, Session};
+
+fn session(artboards: usize) -> Session {
+    let mut s = Session::new();
+    s.execute("file.new", &json!({"width": 100, "height": 80, "artboards": artboards})).unwrap();
+    s.execute("shape.rectangle", &json!({"x": 10, "y": 10, "width": 40, "height": 30})).unwrap();
+    s
+}
+
+fn revision(s: &Session) -> u64 {
+    s.doc().unwrap().revision
+}
+
+#[test]
+fn a_pdf_export_cancelled_between_pages_stops_writes_nothing_and_changes_nothing() {
+    let path = vectorcraft_testkit::temp_dir("cancel-pdf").join("out.pdf");
+    std::fs::remove_file(&path).ok();
+    let mut s = session(5);
+    let before = (revision(&s), s.doc().unwrap().history.undo.len());
+    let token = CancelToken::new();
+    let pages = Rc::new(Cell::new(0));
+    // The sink runs inside the export, as the page loop reports each page: cancel on the third.
+    let sink: ProgressSink = {
+        let (token, pages) = (token.clone(), pages.clone());
+        Rc::new(move |done, total, _| {
+            // Stages before the pages report without a total.
+            if total != 5 {
+                return;
+            }
+            pages.set(done);
+            if done == 3 {
+                token.cancel();
+            }
+        })
+    };
+    let params = json!({"path": path.to_str().unwrap()});
+    let r = cancel::scope(&token, Some(sink), || s.execute("document.exportPdf", &params));
+    assert!(matches!(r, Err(EngineError::Cancelled)), "{r:?}");
+    assert_eq!(pages.get(), 3, "the export stopped at the checkpoint after the cancel, not at the end");
+    assert!(!path.exists(), "a cancelled export writes no file");
+    assert_eq!((revision(&s), s.doc().unwrap().history.undo.len()), before, "the document is as it was");
+    // The session is fine: the same export, not cancelled, completes.
+    s.execute("document.exportPdf", &params).unwrap();
+    assert!(path.exists());
+}
+
+#[test]
+fn a_cancelled_flattening_export_stops_before_the_pages() {
+    let mut s = session(2);
+    let token = CancelToken::new();
+    token.cancel();
+    // PDF/X-1a flattens transparency first; a token raised beforehand stops at the first check.
+    let r = cancel::scope(&token, None, || s.execute("document.exportPdf", &json!({"standard": "pdfX1a"})));
+    assert!(matches!(r, Err(EngineError::Cancelled)), "{r:?}");
+    assert!(s.execute("document.exportPdf", &json!({"standard": "pdfX1a"})).is_ok());
+}
+
+#[test]
+fn raster_exports_and_svg_stop_when_cancelled_too() {
+    let dir = vectorcraft_testkit::temp_dir("cancel-formats");
+    let mut s = session(3);
+    let token = CancelToken::new();
+    token.cancel();
+    for ext in ["png", "svg", "pdf"] {
+        let path = dir.join(format!("out.{ext}"));
+        std::fs::remove_file(&path).ok();
+        let params = json!({"path": path.to_str().unwrap()});
+        let r = cancel::scope(&token, None, || s.execute("document.export", &params));
+        assert!(matches!(r, Err(EngineError::Cancelled)), "{ext}: {r:?}");
+        assert!(!path.exists(), "{ext}: nothing is written");
+    }
+}
+
+#[test]
+fn nothing_is_cancelled_outside_a_scope() {
+    let token = CancelToken::new();
+    token.cancel();
+    // The token only counts for work run under it, on its thread.
+    let mut s = session(1);
+    assert!(s.execute("document.exportPdf", &json!({})).is_ok());
+}

crates/mcp/src/backend.rs

@@ -85,6 +85,42 @@ impl Remote {
         Ok(())
     }
 
+    /// Wait until the app starts to answer, giving up when the call is cancelled
+    /// ([`vectorcraft_engine::cancel`]): the app has no way to stop what it runs, but this server
+    /// stops waiting for it (the connection is dropped, so a late answer can't be mistaken for
+    /// the next call's). Without a cancellable call this is just the read the caller does next.
+    fn wait_for_reply(stream: &TcpStream) -> std::io::Result<()> {
+        const POLL: Duration = Duration::from_millis(100);
+        const LIMIT: Duration = Duration::from_secs(90);
+        if !vectorcraft_engine::cancel::is_cancellable() {
+            return Ok(());
+        }
+        let start = std::time::Instant::now();
+        let kept = stream.read_timeout()?;
+        stream.set_read_timeout(Some(POLL))?;
+        let outcome = loop {
+            match stream.peek(&mut [0]) {
+                // Bytes (or the end of the stream) to read now.
+                Ok(_) => break Ok(()),
+                Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {
+                    if vectorcraft_engine::cancel::is_cancelled() {
+                        break Err(std::io::Error::new(
+                            std::io::ErrorKind::Interrupted,
+                            "cancelled; the app may still finish what it was asked, which can't be stopped from here",
+                        ));
+                    }
+                    if start.elapsed() >= LIMIT {
+                        break Err(e);
+                    }
+                }
+                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
+                Err(e) => break Err(e),
+            }
+        };
+        stream.set_read_timeout(kept)?;
+        outcome
+    }
+
     fn roundtrip(&mut self, line: &str) -> std::io::Result<String> {
         let Some((reader, writer)) = self.conn.as_mut() else {
             return Err(std::io::Error::new(std::io::ErrorKind::NotConnected, "not connected to the app"));
@@ -92,6 +128,7 @@ impl Remote {
         writer.write_all(line.as_bytes())?;
         writer.write_all(b"\n")?;
         writer.flush()?;
+        Self::wait_for_reply(writer)?;
         let mut buf = Vec::new();
         if !crate::server::read_line_capped(reader, &mut buf, MAX_REPLY_BYTES)? {
             return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "control channel closed"));

crates/mcp/src/headless.rs

@@ -253,6 +253,8 @@ impl Headless {
         let scale = p.get("scale").and_then(Value::as_f64).unwrap_or(1.0).clamp(0.01, 16.0);
         vectorcraft_render::raster_size(r, scale)?;
         let img = self.renderer.render_region(&doc, r, scale, true);
+        // A cancelled render stops drawing: the picture is not whole and is not written.
+        vectorcraft_engine::cancel::check().map_err(|e| e.to_string())?;
         let png = img.to_png()?;
         let mut out = json!({"width": img.width, "height": img.height});
         match s(p, "path") {

crates/mcp/src/intake.rs

@@ -0,0 +1,309 @@
+//! What lets the server hear a client while a call runs: a reader thread that takes lines off the
+//! transport, a queue the engine thread works through, and a writer thread that owns the output.
+//!
+//! The engine runs one command at a time, on the thread that called [`crate::Server::serve`]. The
+//! reader never touches the engine. While a call runs it can still
+//!
+//! - act on [`notifications/cancelled`](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation):
+//!   a queued request is dropped, a running one has its [`CancelToken`] raised, and neither gets
+//!   a reply;
+//! - answer the requests that need no engine (`ping`, the `…/list` methods, `prompts/get`), and
+//! - refuse, with a "busy" error that names the running call, a request that would have to wait
+//!   behind a call that has already run for [`Server::with_busy_after`](crate::Server::with_busy_after).
+//!
+//! Requests that arrive sooner than that queue up and run in order, so a client that pipelines a
+//! few quick calls sees them answered in order, as before. The queue is bounded: past
+//! [`MAX_QUEUED`] lines the reader waits for the engine rather than buffering without bound.
+
+use std::collections::VecDeque;
+use std::io::BufRead;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::SyncSender;
+use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
+use std::time::{Duration, Instant};
+
+use serde_json::{Value, json};
+use vectorcraft_engine::cancel::CancelToken;
+
+use crate::server::{INVALID_REQUEST, MAX_LINE_BYTES, PARSE_ERROR, SERVER_BUSY, error, error_with, read_line_capped};
+
+/// Most lines waiting for the engine. Past this the reader waits (it still reads nothing else).
+pub(crate) const MAX_QUEUED: usize = 64;
+
+/// A line the engine thread has still to run.
+pub(crate) enum Job {
+    /// A reply that is already known (a parse error, an overlong line), kept in order with the rest.
+    Canned(String),
+    /// A parsed message: a request, a notification or a batch.
+    Message(Message),
+}
+
+pub(crate) struct Message {
+    pub(crate) value: Value,
+    /// The request id as JSON text, for a request; what `notifications/cancelled` names.
+    pub(crate) key: Option<String>,
+    pub(crate) method: String,
+    /// Raised by `notifications/cancelled`, or when the transport closes under the call.
+    pub(crate) token: CancelToken,
+    /// What the call is, for the "busy" answer: `tools/call command_run document.export`.
+    pub(crate) label: String,
+}
+
+/// The call the engine thread is running.
+struct Running {
+    key: Option<String>,
+    method: String,
+    token: CancelToken,
+    label: String,
+    since: Instant,
+}
+
+#[derive(Default)]
+struct State {
+    queue: VecDeque<Job>,
+    running: Option<Running>,
+    /// The transport has no more lines.
+    eof: bool,
+    /// The output failed: nobody reads replies any more.
+    closed: bool,
+}
+
+/// What the reader and the engine thread share.
+pub(crate) struct Shared {
+    state: Mutex<State>,
+    wake: Condvar,
+    /// Whether the client speaks the `2026-07-28` revision, for results answered by the reader.
+    pub(crate) modern: Arc<AtomicBool>,
+    busy_after: Duration,
+}
+
+fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
+    // A thread that panicked while holding the lock left a queue that is still well formed.
+    m.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+impl Shared {
+    pub(crate) fn new(modern: Arc<AtomicBool>, busy_after: Duration) -> Self {
+        Self { state: Mutex::new(State::default()), wake: Condvar::new(), modern, busy_after }
+    }
+
+    /// The next job, or `None` once the transport is done and the queue empty (or the output
+    /// failed). The job counts as running until [`Shared::finished`].
+    pub(crate) fn next(&self) -> Option<Job> {
+        let mut st = lock(&self.state);
+        loop {
+            if st.closed {
+                return None;
+            }
+            if let Some(job) = st.queue.pop_front() {
+                if let Job::Message(m) = &job {
+                    st.running = Some(Running {
+                        key: m.key.clone(),
+                        method: m.method.clone(),
+                        token: m.token.clone(),
+                        label: m.label.clone(),
+                        since: Instant::now(),
+                    });
+                }
+                // The reader may be waiting for room.
+                self.wake.notify_all();
+                return Some(job);
+            }
+            if st.eof {
+                return None;
+            }
+            st = self.wake.wait(st).unwrap_or_else(PoisonError::into_inner);
+        }
+    }
+
+    pub(crate) fn finished(&self) {
+        lock(&self.state).running = None;
+        self.wake.notify_all();
+    }
+
+    /// The output failed: stop the running call and everything queued behind it.
+    pub(crate) fn close(&self) {
+        let mut st = lock(&self.state);
+        st.closed = true;
+        if let Some(r) = &st.running {
+            r.token.cancel();
+        }
+        st.queue.clear();
+        self.wake.notify_all();
+    }
+
+    fn end_of_input(&self) {
+        lock(&self.state).eof = true;
+        self.wake.notify_all();
+    }
+
+    fn is_closed(&self) -> bool {
+        lock(&self.state).closed
+    }
+
+    /// The running call, once it has run for `busy_after` (so a quick call never makes a
+    /// pipelined neighbour fail).
+    fn long_running(&self, st: &State) -> Option<Value> {
+        let r = st.running.as_ref()?;
+        let secs = r.since.elapsed();
+        (secs >= self.busy_after).then(|| json!({"running": r.label, "requestId": r.key, "seconds": secs.as_secs()}))
+    }
+}
+
+/// The id of a request message: `Some` when `msg` is a JSON-RPC 2.0 request with a usable id.
+fn request_id(msg: &Value) -> Option<&Value> {
+    let o = msg.as_object()?;
+    if o.get("jsonrpc").and_then(Value::as_str) != Some("2.0") || !o.get("method")?.is_string() {
+        return None;
+    }
+    o.get("id").filter(|id| id.is_string() || id.is_number())
+}
+
+/// A short name for the call, as the "busy" answer shows it.
+fn label_of(method: &str, params: &Value) -> String {
+    if method != "tools/call" {
+        return method.to_string();
+    }
+    let name = params.get("name").and_then(Value::as_str).unwrap_or("?");
+    let args = params.get("arguments");
+    let which = ["id", "command", "path"].iter().find_map(|k| args?.get(k)?.as_str());
+    match which {
+        Some(w) => format!("{method} {name} {w}"),
+        None => format!("{method} {name}"),
+    }
+}
+
+/// Read lines from `reader` until the end, acting on each as described in the module docs.
+pub(crate) fn read_loop(shared: &Shared, out: &SyncSender<String>, mut reader: impl BufRead) -> std::io::Result<()> {
+    let mut buf = Vec::new();
+    let result = loop {
+        match read_line_capped(&mut reader, &mut buf, MAX_LINE_BYTES) {
+            Ok(true) => {}
+            Ok(false) => break Ok(()),
+            Err(e) => break Err(e),
+        }
+        if shared.is_closed() {
+            break Ok(());
+        }
+        if buf.len() > MAX_LINE_BYTES {
+            // Longer than anything a client can mean: answer the framing violation, skip the
+            // rest of the line in bounded chunks, and keep serving what follows.
+            let reply = error(Value::Null, INVALID_REQUEST, format!("request line longer than {MAX_LINE_BYTES} bytes")).to_string();
+            submit_canned(shared, out, reply);
+            let ended = loop {
+                match read_line_capped(&mut reader, &mut buf, MAX_LINE_BYTES) {
+                    Ok(true) if buf.len() > MAX_LINE_BYTES => {}
+                    Ok(true) => break Ok(false),
+                    Ok(false) => break Ok(true),
+                    Err(e) => break Err(e),
+                }
+            };
+            match ended {
+                Ok(false) => continue,
+                Ok(true) => break Ok(()),
+                Err(e) => break Err(e),
+            }
+        }
+        let line = match std::str::from_utf8(&buf) {
+            Ok(l) => l,
+            Err(e) => break Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
+        };
+        intake(shared, out, line);
+    };
+    shared.end_of_input();
+    result
+}
+
+/// A reply that is already known: in order behind queued work, or at once when the engine is
+/// tied up for long (then nobody can expect order).
+fn submit_canned(shared: &Shared, out: &SyncSender<String>, reply: String) {
+    let mut st = lock(&shared.state);
+    if shared.long_running(&st).is_some() {
+        drop(st);
+        let _ = out.send(reply);
+        return;
+    }
+    st.queue.push_back(Job::Canned(reply));
+    shared.wake.notify_all();
+}
+
+fn intake(shared: &Shared, out: &SyncSender<String>, line: &str) {
+    let line = line.trim();
+    if line.is_empty() {
+        return;
+    }
+    let value = match serde_json::from_str::<Value>(line) {
+        Ok(v) => v,
+        Err(e) => return submit_canned(shared, out, error(Value::Null, PARSE_ERROR, format!("parse error: {e}")).to_string()),
+    };
+    let method = value.get("method").and_then(Value::as_str).unwrap_or("").to_string();
+    let enveloped = value.get("jsonrpc").and_then(Value::as_str) == Some("2.0");
+    if enveloped && method == "notifications/cancelled" && value.get("id").is_none() {
+        return cancel(shared, value.get("params").unwrap_or(&Value::Null));
+    }
+    let params = value.get("params").cloned().unwrap_or(Value::Null);
+    let id = request_id(&value).cloned();
+    let is_batch = value.is_array();
+    let token = CancelToken::new();
+    let msg = Message { key: id.as_ref().map(Value::to_string), label: label_of(&method, &params), method, token, value };
+    let mut st = lock(&shared.state);
+    loop {
+        if st.closed {
+            return;
+        }
+        let busy = shared.long_running(&st);
+        if let Some(busy) = busy {
+            if let Some(id) = &id {
+                // Engine-free requests are answered here; others are told to retry.
+                let reply = match crate::server::stateless_reply(&msg.value, shared.modern.load(Ordering::Relaxed)) {
+                    Some(r) => r,
+                    None => error_with(id.clone(), SERVER_BUSY, busy_message(&busy), json!({"busy": busy})),
+                };
+                drop(st);
+                let _ = out.send(reply.to_string());
+                return;
+            }
+            if is_batch {
+                drop(st);
+                let reply = error_with(Value::Null, SERVER_BUSY, busy_message(&busy), json!({"busy": busy}));
+                let _ = out.send(reply.to_string());
+                return;
+            }
+        }
+        if st.queue.len() < MAX_QUEUED {
+            break;
+        }
+        // Wake now and then: the call that fills the queue may be the one that turns it away.
+        st = shared.wake.wait_timeout(st, Duration::from_millis(50)).unwrap_or_else(PoisonError::into_inner).0;
+    }
+    st.queue.push_back(Job::Message(msg));
+    shared.wake.notify_all();
+}
+
+fn busy_message(busy: &Value) -> String {
+    format!(
+        "the engine is busy: `{}` has been running for {} s and this request was not run. Retry when it has finished, or cancel it with notifications/cancelled (requestId {})",
+        busy["running"].as_str().unwrap_or("a call"),
+        busy["seconds"].as_u64().unwrap_or(0),
+        busy["requestId"],
+    )
+}
+
+/// `notifications/cancelled {requestId}`: stop that request if it is still queued or running.
+/// An unknown or finished id is ignored, as is a request that can't be cancelled (`initialize`).
+fn cancel(shared: &Shared, params: &Value) {
+    let Some(id) = params.get("requestId").filter(|id| id.is_string() || id.is_number()) else { return };
+    let key = id.to_string();
+    let mut st = lock(&shared.state);
+    if let Some(r) = st.running.as_ref().filter(|r| r.key.as_deref() == Some(&key) && r.method != "initialize") {
+        log::debug!("cancelling running request {key}: {}", r.label);
+        r.token.cancel();
+        return;
+    }
+    let at = st.queue.iter().position(|j| matches!(j, Job::Message(m) if m.key.as_deref() == Some(&key) && m.method != "initialize"));
+    if let Some(at) = at {
+        log::debug!("dropping queued request {key}");
+        st.queue.remove(at);
+        shared.wake.notify_all();
+    }
+}

crates/mcp/src/lib.rs

@@ -16,6 +16,7 @@
 
 mod backend;
 mod headless;
+mod intake;
 pub mod logging;
 mod prompts;
 mod resources;
@@ -37,6 +38,8 @@ mod tests;
 #[cfg(test)]
 mod tests_batchdraw;
 #[cfg(test)]
+mod tests_concurrency;
+#[cfg(test)]
 mod tests_conventions;
 #[cfg(test)]
 mod tests_distortkeys;

crates/mcp/src/server.rs

@@ -1,10 +1,18 @@
 //! JSON-RPC 2.0 framing and the MCP lifecycle / tools / resources / prompts methods.
 
+use std::cell::Cell;
 use std::io::{BufRead, Write};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::{SyncSender, sync_channel};
+use std::time::{Duration, Instant};
 
 use serde_json::{Value, json};
+use vectorcraft_engine::cancel;
 
 use crate::backend::Backend;
+use crate::intake::{Job, Message, Shared, read_loop};
 use crate::prompts;
 use crate::resources;
 use crate::tools::{call_tool, tool_definitions};
@@ -14,12 +22,18 @@ pub const PROTOCOL_VERSION: &str = "2025-06-18";
 const MODERN_VERSION: &str = "2026-07-28";
 const SUPPORTED_VERSIONS: &[&str] = &[MODERN_VERSION, PROTOCOL_VERSION, "2025-03-26", "2024-11-05"];
 
-const PARSE_ERROR: i64 = -32700;
-const INVALID_REQUEST: i64 = -32600;
+pub(crate) const PARSE_ERROR: i64 = -32700;
+pub(crate) const INVALID_REQUEST: i64 = -32600;
 const METHOD_NOT_FOUND: i64 = -32601;
 const INVALID_PARAMS: i64 = -32602;
 const INTERNAL_ERROR: i64 = -32603;
 const RESOURCE_NOT_FOUND: i64 = -32002;
+/// The engine is running a long call and this request was not run (implementation-defined range).
+pub(crate) const SERVER_BUSY: i64 = -32003;
+
+/// How long a call runs before a request that would have to wait behind it is refused as busy
+/// instead ([`Server::with_busy_after`]).
+const DEFAULT_BUSY_AFTER: Duration = Duration::from_secs(2);
 
 /// Longest accepted request line (bytes, without the newline). Requests are small JSON objects;
 /// anything longer is answered with an error and skipped, so one huge line can't make the server
@@ -29,6 +43,40 @@ pub(crate) const MAX_LINE_BYTES: usize = 16 * 1024 * 1024;
 /// older clients can make the server parse and answer at once (#1727).
 pub(crate) const MAX_BATCH_MESSAGES: usize = 64;
 
+/// Lines waiting to be written. A full queue holds up the thread that wants to write (the
+/// engine thread, the reader) rather than growing; progress is dropped instead (it is advisory).
+const OUTBOX: usize = 256;
+/// Fewest time between two `notifications/progress` of one request, the last one excepted.
+const PROGRESS_EVERY: Duration = Duration::from_millis(200);
+
+/// The sink that turns the engine's progress into `notifications/progress` for the request
+/// `msg`, when the client asked for it with `_meta.progressToken`.
+/// <https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress>
+fn progress_sink(out: SyncSender<String>, msg: &Value) -> Option<cancel::ProgressSink> {
+    let token = msg.get("params")?.get("_meta")?.get("progressToken").filter(|t| t.is_string() || t.is_number())?.clone();
+    let last = Cell::new(None::<u64>);
+    let at = Cell::new(None::<Instant>);
+    Some(Rc::new(move |done, total, what| {
+        // `progress` must increase with each notification.
+        if last.get().is_some_and(|l| done <= l) {
+            return;
+        }
+        let finished = total > 0 && done >= total;
+        if !finished && at.get().is_some_and(|t| t.elapsed() < PROGRESS_EVERY) {
+            return;
+        }
+        last.set(Some(done));
+        at.set(Some(Instant::now()));
+        let mut params = json!({"progressToken": token, "progress": done, "message": what});
+        if total > 0 {
+            params["total"] = json!(total);
+        }
+        let line = json!({"jsonrpc": "2.0", "method": "notifications/progress", "params": params}).to_string();
+        // Advisory: a client that doesn't read fast enough misses an update, not the reply.
+        let _ = out.try_send(line);
+    }))
+}
+
 const INSTRUCTIONS: &str = "VectorCraft is a professional vector illustration app. Coordinates are \
 points in document space (y down, origin at the first artboard's top-left; a new document is 612×792). \
 Draw with draw_shape / draw_path, change colours with set_paint, look with render_preview and doc_inspect. \
@@ -42,20 +90,88 @@ vectorcraft://swatch/{name} read one thing at a time instead of the whole docume
 pub struct Server {
     backend: Box<dyn Backend>,
     initialized: bool,
-    modern: bool,
+    /// Whether the client speaks the `2026-07-28` revision. Shared with the reader thread, which
+    /// answers some requests itself while the engine is busy.
+    modern: Arc<AtomicBool>,
     /// Whether this server's client asked for log records (`logging/setLevel`). The queue is
     /// process-wide, so a server whose client didn't ask leaves it to the one that did.
     logging: bool,
+    busy_after: Duration,
 }
 
-fn response(id: Value, result: Value) -> Value {
+pub(crate) fn response(id: Value, result: Value) -> Value {
     json!({"jsonrpc": "2.0", "id": id, "result": result})
 }
 
-fn error(id: Value, code: i64, message: impl Into<String>) -> Value {
+pub(crate) fn error(id: Value, code: i64, message: impl Into<String>) -> Value {
     json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message.into()}})
 }
 
+pub(crate) fn error_with(id: Value, code: i64, message: impl Into<String>, data: Value) -> Value {
+    json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message.into(), "data": data}})
+}
+
+/// The requests that need neither the engine nor the server's state beyond the protocol revision,
+/// so the reader can answer them while the engine is busy.
+pub(crate) fn stateless_request(method: &str, params: &Value) -> Option<Result<Value, (i64, String)>> {
+    Some(match method {
+        "ping" => Ok(json!({})),
+        "tools/list" => Ok(json!({"tools": tool_definitions()})),
+        "resources/list" => Ok(resources::list()),
+        "resources/templates/list" => Ok(resources::templates()),
+        "prompts/list" => Ok(prompts::list()),
+        "prompts/get" => {
+            let name = params.get("name").and_then(Value::as_str);
+            let args = params.get("arguments").cloned().unwrap_or(Value::Null);
+            match name {
+                Some(name) => prompts::get(name, &args).map_err(|e| (INVALID_PARAMS, e)),
+                None => Err((INVALID_PARAMS, "missing prompt `name`".to_string())),
+            }
+        }
+        _ => return None,
+    })
+}
+
+/// The reply to `msg` when it is a request [`stateless_request`] answers.
+pub(crate) fn stateless_reply(msg: &Value, modern: bool) -> Option<Value> {
+    let o = msg.as_object()?;
+    let method = o.get("method")?.as_str()?;
+    let id = o.get("id").filter(|id| id.is_string() || id.is_number())?.clone();
+    if o.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
+        return None;
+    }
+    let params = o.get("params").cloned().unwrap_or(Value::Null);
+    // A panic here is a bug: leave the request to the engine thread, which reports it.
+    let r = vectorcraft_engine::guard::catch_panic(|| stateless_request(method, &params)).ok()??;
+    Some(match r {
+        Ok(r) => response(id, complete_result(method, &params, modern, r)),
+        Err((code, m)) => error(id, code, m),
+    })
+}
+
+/// The cache hints the `2026-07-28` revision adds to list and read results. `modern` is whether the
+/// session negotiated that revision; a request can also name it in its `_meta`.
+pub(crate) fn complete_result(method: &str, params: &Value, modern: bool, mut r: Value) -> Value {
+    let modern = params
+        .get("_meta")
+        .and_then(|m| m.get("io.modelcontextprotocol/protocolVersion"))
+        .and_then(Value::as_str)
+        .map_or(modern, |v| v == MODERN_VERSION);
+    if modern && let Some(result) = r.as_object_mut() {
+        let ttl = match method {
+            "tools/list" | "resources/list" | "resources/templates/list" | "prompts/list" => Some(600_000),
+            "resources/read" | "prompts/get" => Some(0),
+            _ => None,
+        };
+        if let Some(ttl) = ttl {
+            result.insert("resultType".into(), json!("complete"));
+            result.insert("ttlMs".into(), json!(ttl));
+            result.insert("cacheScope".into(), json!("private"));
+        }
+    }
+    r
+}
+
 /// Read one `\n`-terminated line into `buf` (newline stripped). `Ok(false)` at end of stream.
 /// `buf` holding more than `max` bytes means the line is overlong: no newline arrived within the
 /// bound, so the bytes so far are only its start. Only `max + 1` bytes are ever buffered for one
@@ -101,7 +217,15 @@ fn write_reply(output: &mut impl Write, line: &str) -> std::io::Result<()> {
 
 impl Server {
     pub fn new(backend: Box<dyn Backend>) -> Self {
-        Self { backend, initialized: false, modern: false, logging: false }
+        Self { backend, initialized: false, modern: Arc::new(AtomicBool::new(false)), logging: false, busy_after: DEFAULT_BUSY_AFTER }
+    }
+
+    /// How long a call runs before [`Server::serve`] turns away a request that would have to wait
+    /// behind it, with an error naming the call (default two seconds). Shorter waits queue, so a
+    /// client pipelining quick calls is answered in order.
+    pub fn with_busy_after(mut self, after: Duration) -> Self {
+        self.busy_after = after;
+        self
     }
 
     pub fn backend(&mut self) -> &mut dyn Backend {
@@ -130,36 +254,76 @@ 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<()> {
-        let mut reader = input;
-        let mut buf = Vec::new();
-        loop {
-            if !read_line_capped(&mut reader, &mut buf, MAX_LINE_BYTES)? {
-                return Ok(());
-            }
-            if buf.len() > MAX_LINE_BYTES {
-                // Longer than anything a client can mean: answer the framing violation, skip the
-                // rest of the line in bounded chunks, and keep serving what follows.
-                write_reply(
-                    &mut output,
-                    &error(Value::Null, INVALID_REQUEST, format!("request line longer than {MAX_LINE_BYTES} bytes")).to_string(),
-                )?;
-                loop {
-                    if !read_line_capped(&mut reader, &mut buf, MAX_LINE_BYTES)? {
-                        return Ok(());
-                    }
-                    if buf.len() <= MAX_LINE_BYTES {
-                        break;
+    ///
+    /// A reader thread takes lines off `input` while the calling thread runs them, so the server
+    /// hears a client during a long call: `notifications/cancelled` stops a queued or running
+    /// request (which then gets no reply), and `ping` and the other engine-free requests are
+    /// answered at once. A request that would have to wait behind a call that has run for
+    /// [`Server::with_busy_after`] is refused with a "busy" error. Calls still run one at a time,
+    /// in arrival order, and replies of quick calls keep that order. See the module docs of
+    /// `intake` and `docs/mcp.md`.
+    ///
+    /// Returns when `input` ends and everything read has been answered, or when `output` fails
+    /// (the running call is then cancelled and the queue dropped; the reader thread ends with its
+    /// next line, which a client that closed our output closes as well).
+    pub fn serve<R: BufRead + Send, W: Write + Send>(&mut self, input: R, mut output: W) -> std::io::Result<()> {
+        let shared = Shared::new(self.modern.clone(), self.busy_after);
+        let shared = &shared;
+        let (tx, rx) = sync_channel::<String>(OUTBOX);
+        let reader_tx = tx.clone();
+        std::thread::scope(|scope| {
+            // The only thread that writes: replies, progress and log records stay whole lines.
+            let writer = scope.spawn(move || {
+                let mut failed = None;
+                for line in rx {
+                    if failed.is_none()
+                        && let Err(e) = write_reply(&mut output, &line)
+                    {
+                        failed = Some(e);
+                        shared.close();
                     }
                 }
-                continue;
+                failed
+            });
+            let reader = scope.spawn(move || read_loop(shared, &reader_tx, input));
+            self.work(shared, &tx);
+            drop(tx);
+            let read = reader.join().unwrap_or_else(|_| Err(std::io::Error::other("the MCP reader thread panicked")));
+            let written = writer.join().unwrap_or_else(|_| Some(std::io::Error::other("the MCP writer thread panicked")));
+            match written {
+                Some(e) => Err(e),
+                None => read,
             }
-            let line = std::str::from_utf8(&buf).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
-            let reply = self.handle_line(line);
-            let notes = self.take_notifications();
-            for line in reply.iter().chain(notes.iter()) {
-                write_reply(&mut output, line)?;
+        })
+    }
+
+    /// Run the queued jobs, one at a time, until the reader is done and the queue empty.
+    fn work(&mut self, shared: &Shared, out: &SyncSender<String>) {
+        while let Some(job) = shared.next() {
+            match job {
+                Job::Canned(line) => {
+                    let _ = out.send(line);
+                }
+                Job::Message(m) => self.run(m, out),
             }
+            for note in self.take_notifications() {
+                let _ = out.send(note);
+            }
+            shared.finished();
+        }
+    }
+
+    fn run(&mut self, m: Message, out: &SyncSender<String>) {
+        let Message { value, key, token, .. } = m;
+        let sink = key.is_some().then(|| progress_sink(out.clone(), &value)).flatten();
+        let reply = cancel::scope(&token, sink, || self.handle_parsed(value));
+        // A cancelled request gets no reply (MCP: the receiver should not answer it). A call that
+        // finished before it noticed is dropped the same way: the client has stopped waiting.
+        if key.is_some() && token.is_cancelled() {
+            return;
+        }
+        if let Some(reply) = reply {
+            let _ = out.send(reply.to_string());
         }
     }
 
@@ -170,7 +334,16 @@ impl Server {
             return None;
         }
         let reply = match serde_json::from_str::<Value>(line) {
-            Ok(Value::Array(batch)) => {
+            Ok(value) => self.handle_parsed(value),
+            Err(e) => Some(error(Value::Null, PARSE_ERROR, format!("parse error: {e}"))),
+        };
+        reply.map(|r| r.to_string())
+    }
+
+    /// Handle one parsed line: a message, or a batch of them.
+    fn handle_parsed(&mut self, value: Value) -> Option<Value> {
+        match value {
+            Value::Array(batch) => {
                 // Batches were removed in 2025-06-18; still answer older clients sensibly.
                 if batch.is_empty() {
                     Some(error(Value::Null, INVALID_REQUEST, "empty batch"))
@@ -181,10 +354,8 @@ impl Server {
                     (!replies.is_empty()).then_some(Value::Array(replies))
                 }
             }
-            Ok(msg) => self.handle(msg),
-            Err(e) => Some(error(Value::Null, PARSE_ERROR, format!("parse error: {e}"))),
-        };
-        reply.map(|r| r.to_string())
+            msg => self.handle(msg),
+        }
     }
 
     /// Handle one JSON-RPC message; `None` for notifications and responses.
@@ -216,26 +387,7 @@ impl Server {
         let r = vectorcraft_engine::guard::catch_panic(|| self.request(method, &params))
             .unwrap_or_else(|msg| Err((INTERNAL_ERROR, format!("internal error in `{method}`: {msg} (please report this bug)"))));
         Some(match r {
-            Ok(mut r) => {
-                let modern = params
-                    .get("_meta")
-                    .and_then(|m| m.get("io.modelcontextprotocol/protocolVersion"))
-                    .and_then(Value::as_str)
-                    .map_or(self.modern, |v| v == MODERN_VERSION);
-                if modern && let Some(result) = r.as_object_mut() {
-                    let ttl = match method {
-                        "tools/list" | "resources/list" | "resources/templates/list" | "prompts/list" => Some(600_000),
-                        "resources/read" | "prompts/get" => Some(0),
-                        _ => None,
-                    };
-                    if let Some(ttl) = ttl {
-                        result.insert("resultType".into(), json!("complete"));
-                        result.insert("ttlMs".into(), json!(ttl));
-                        result.insert("cacheScope".into(), json!("private"));
-                    }
-                }
-                response(id, r)
-            }
+            Ok(r) => response(id, complete_result(method, &params, self.modern.load(Ordering::Relaxed), r)),
             Err((code, m)) => error(id, code, m),
         })
     }
@@ -249,11 +401,14 @@ impl Server {
     }
 
     fn request(&mut self, method: &str, params: &Value) -> Result<Value, (i64, String)> {
+        if let Some(r) = stateless_request(method, params) {
+            return r;
+        }
         match method {
             "initialize" => {
                 let asked = params.get("protocolVersion").and_then(Value::as_str).unwrap_or(PROTOCOL_VERSION);
                 let version = if SUPPORTED_VERSIONS.contains(&asked) { asked } else { PROTOCOL_VERSION };
-                self.modern = version == MODERN_VERSION;
+                self.modern.store(version == MODERN_VERSION, Ordering::Relaxed);
                 Ok(json!({
                     "protocolVersion": version,
                     "capabilities": {
@@ -267,8 +422,6 @@ impl Server {
                     "instructions": format!("{INSTRUCTIONS} Backend: {}.", self.backend.describe()),
                 }))
             }
-            "ping" => Ok(json!({})),
-            "tools/list" => Ok(json!({"tools": tool_definitions()})),
             "tools/call" => {
                 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);
@@ -277,8 +430,6 @@ impl Server {
                 }
                 Ok(call_tool(self.backend.as_mut(), name, &args).to_value())
             }
-            "resources/list" => Ok(resources::list()),
-            "resources/templates/list" => Ok(resources::templates()),
             "resources/read" => {
                 let uri = params.get("uri").and_then(Value::as_str).ok_or((INVALID_PARAMS, "missing `uri`".to_string()))?;
                 let v = resources::read(self.backend.as_mut(), uri).map_err(|e| match e {
@@ -288,12 +439,6 @@ impl Server {
                 })?;
                 Ok(json!({"contents": [{"uri": uri, "mimeType": "application/json", "text": serde_json::to_string_pretty(&v).unwrap_or_default()}]}))
             }
-            "prompts/list" => Ok(prompts::list()),
-            "prompts/get" => {
-                let name = params.get("name").and_then(Value::as_str).ok_or((INVALID_PARAMS, "missing prompt `name`".to_string()))?;
-                let args = params.get("arguments").cloned().unwrap_or(Value::Null);
-                prompts::get(name, &args).map_err(|e| (INVALID_PARAMS, e))
-            }
             "completion/complete" => Ok(prompts::complete(self.backend.as_mut(), params)),
             "logging/setLevel" => {
                 // `level` is required. A null level is our own way of turning logging back off,

crates/mcp/src/tests_concurrency.rs

@@ -0,0 +1,368 @@
+//! Reading while a call runs: `ping`, cancellation and busy answers reach the client while a long
+//! command holds the engine (#2165, #2171). Driven through [`Server::serve`] on a pipe, the way a
+//! client on stdio drives it, with a backend that works until it is told to stop.
+
+use std::io::{BufReader, Write};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use serde_json::{Value, json};
+use vectorcraft_engine::cancel;
+
+use crate::{Backend, Server};
+
+/// A backend whose `engine.execute` runs until cancelled (or for `limit`), like a long export.
+struct Slow {
+    limit: Duration,
+    /// Commands that ran to the end, and the ones that were cancelled.
+    log: Arc<Mutex<Vec<String>>>,
+    /// Told when a command starts.
+    started: Sender<String>,
+}
+
+impl Backend for Slow {
+    fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
+        if method != "engine.execute" {
+            return Ok(json!({"method": method}));
+        }
+        let id = params.get("command").and_then(Value::as_str).unwrap_or("?").to_string();
+        self.started.send(id.clone()).ok();
+        if id == "progress.demo" {
+            for page in 1..=3 {
+                // Enough apart to pass the sender's throttle.
+                std::thread::sleep(Duration::from_millis(230));
+                cancel::progress(page, 3, &format!("page {page}"));
+            }
+            return Ok(json!({"ran": id}));
+        }
+        let limit = if id.starts_with("quick") { Duration::ZERO } else { self.limit };
+        let start = Instant::now();
+        while start.elapsed() < limit {
+            if let Err(e) = cancel::check() {
+                self.log.lock().unwrap().push(format!("cancelled {id}"));
+                return Err(e.to_string());
+            }
+            std::thread::sleep(Duration::from_millis(5));
+        }
+        self.log.lock().unwrap().push(format!("done {id}"));
+        Ok(json!({"ran": id}))
+    }
+
+    fn has_ui(&self) -> bool {
+        false
+    }
+
+    fn describe(&self) -> String {
+        "slow".into()
+    }
+}
+
+/// Collects the lines the server writes.
+struct Lines(Sender<String>, Vec<u8>);
+
+impl Write for Lines {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        for &b in buf {
+            if b == b'\n' {
+                self.0.send(String::from_utf8_lossy(&std::mem::take(&mut self.1)).into_owned()).ok();
+            } else {
+                self.1.push(b);
+            }
+        }
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+/// A running server with a pipe in and a channel out.
+struct Rig {
+    to_server: std::io::PipeWriter,
+    replies: Receiver<String>,
+    started: Receiver<String>,
+    log: Arc<Mutex<Vec<String>>>,
+    thread: Option<std::thread::JoinHandle<()>>,
+}
+
+impl Rig {
+    fn new(limit: Duration, busy_after: Duration) -> Self {
+        let (read, to_server) = std::io::pipe().unwrap();
+        let (tx, replies) = mpsc::channel();
+        let (started_tx, started) = mpsc::channel();
+        let log = Arc::new(Mutex::new(vec![]));
+        let backend = Slow { limit, log: log.clone(), started: started_tx };
+        let thread = std::thread::spawn(move || {
+            let mut server = Server::new(Box::new(backend)).with_busy_after(busy_after);
+            server.serve(BufReader::new(read), Lines(tx, vec![])).unwrap();
+        });
+        Self { to_server, replies, started, log, thread: Some(thread) }
+    }
+
+    fn send(&mut self, msg: Value) {
+        writeln!(self.to_server, "{msg}").unwrap();
+    }
+
+    fn request(&mut self, id: u64, method: &str, params: Value) {
+        self.send(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}));
+    }
+
+    fn run(&mut self, id: u64, command: &str) {
+        self.request(id, "tools/call", json!({"name": "command_run", "arguments": {"id": command}}));
+    }
+
+    fn notify(&mut self, method: &str, params: Value) {
+        self.send(json!({"jsonrpc": "2.0", "method": method, "params": params}));
+    }
+
+    fn next(&self, within: Duration) -> Option<Value> {
+        match self.replies.recv_timeout(within) {
+            Ok(l) => serde_json::from_str(&l).ok(),
+            Err(RecvTimeoutError::Timeout) => None,
+            Err(RecvTimeoutError::Disconnected) => None,
+        }
+    }
+
+    /// The next reply that answers request `id`, skipping notifications.
+    fn reply(&self, id: u64, within: Duration) -> Option<Value> {
+        let end = Instant::now() + within;
+        while let Some(left) = end.checked_duration_since(Instant::now()) {
+            let v = self.next(left)?;
+            if v.get("id") == Some(&json!(id)) {
+                return Some(v);
+            }
+        }
+        None
+    }
+
+    fn finish(mut self) {
+        let thread = self.thread.take().unwrap();
+        drop(self.to_server);
+        thread.join().unwrap();
+    }
+}
+
+const LONG: Duration = Duration::from_secs(20);
+const QUICK: Duration = Duration::from_secs(3);
+
+#[test]
+fn ping_is_answered_while_a_command_runs() {
+    let mut rig = Rig::new(LONG, Duration::from_millis(50));
+    rig.run(1, "document.export");
+    assert_eq!(rig.started.recv_timeout(QUICK).unwrap(), "document.export");
+    // Past `busy_after` the call counts as long-running.
+    std::thread::sleep(Duration::from_millis(120));
+    rig.request(2, "ping", json!({}));
+    let pong = rig.reply(2, QUICK).expect("ping must not wait for the running command");
+    assert_eq!(pong["result"], json!({}));
+    rig.notify("notifications/cancelled", json!({"requestId": 1}));
+    rig.finish();
+}
+
+#[test]
+fn cancelling_stops_the_running_command_and_sends_no_reply() {
+    let mut rig = Rig::new(LONG, Duration::from_secs(60));
+    rig.run(1, "document.export");
+    rig.started.recv_timeout(QUICK).unwrap();
+    rig.notify("notifications/cancelled", json!({"requestId": 1, "reason": "user"}));
+    // The engine is free again: the next command runs and answers.
+    rig.run(2, "pdf.preset.list");
+    let end = Instant::now() + QUICK;
+    loop {
+        if rig.log.lock().unwrap().iter().any(|l| l == "cancelled document.export") {
+            break;
+        }
+        assert!(Instant::now() < end, "the export kept running: {:?}", rig.log.lock().unwrap());
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    rig.notify("notifications/cancelled", json!({"requestId": 2}));
+    // Request 1 was cancelled: nothing answers it (MCP: the receiver should not reply).
+    assert!(rig.reply(1, Duration::from_millis(300)).is_none());
+    rig.finish();
+}
+
+#[test]
+fn a_request_that_would_wait_behind_a_long_call_is_refused_as_busy() {
+    let mut rig = Rig::new(LONG, Duration::from_millis(50));
+    rig.run(1, "document.export");
+    rig.started.recv_timeout(QUICK).unwrap();
+    std::thread::sleep(Duration::from_millis(120));
+    rig.run(2, "pdf.preset.list");
+    let busy = rig.reply(2, QUICK).expect("a busy answer, not a wait");
+    assert_eq!(busy["error"]["code"], -32003, "{busy}");
+    assert!(busy["error"]["message"].as_str().unwrap().contains("document.export"), "{busy}");
+    assert_eq!(busy["error"]["data"]["busy"]["requestId"], "1", "{busy}");
+    rig.notify("notifications/cancelled", json!({"requestId": 1}));
+    // The refused request never ran, and the engine is free once the export stops.
+    rig.run(3, "pdf.preset.list");
+    rig.notify("notifications/cancelled", json!({"requestId": 3}));
+    rig.finish();
+}
+
+#[test]
+fn quick_calls_pipelined_together_queue_and_answer_in_order() {
+    let mut rig = Rig::new(Duration::from_millis(30), Duration::from_secs(60));
+    for id in 1..=4 {
+        rig.run(id, &format!("cmd{id}"));
+    }
+    for id in 1..=4 {
+        let r = rig.reply(id, QUICK).unwrap_or_else(|| panic!("no reply to {id}"));
+        assert!(r.get("error").is_none(), "{r}");
+    }
+    assert_eq!(*rig.log.lock().unwrap(), ["done cmd1", "done cmd2", "done cmd3", "done cmd4"]);
+    rig.finish();
+}
+
+#[test]
+fn a_queued_request_that_is_cancelled_never_runs() {
+    let mut rig = Rig::new(LONG, Duration::from_secs(60));
+    rig.run(1, "document.export");
+    rig.started.recv_timeout(QUICK).unwrap();
+    rig.run(2, "never.runs");
+    rig.run(3, "quick.last");
+    // Request 2 may not have been read yet when the cancel is: it is the order of lines that counts.
+    rig.notify("notifications/cancelled", json!({"requestId": 2}));
+    rig.notify("notifications/cancelled", json!({"requestId": 1}));
+    let r = rig.reply(3, QUICK).expect("the engine goes on to the next request");
+    assert!(r.get("error").is_none(), "{r}");
+    rig.notify("notifications/cancelled", json!({"requestId": 3}));
+    let log = rig.log.lock().unwrap().clone();
+    assert!(log.iter().all(|l| !l.contains("never.runs")), "{log:?}");
+    assert!(log.iter().any(|l| l == "cancelled document.export"), "{log:?}");
+    drop(log);
+    rig.finish();
+}
+
+#[test]
+fn cancelling_an_unknown_or_finished_request_changes_nothing() {
+    let mut rig = Rig::new(Duration::from_millis(10), Duration::from_secs(60));
+    rig.run(1, "quick");
+    assert!(rig.reply(1, QUICK).unwrap().get("result").is_some());
+    rig.notify("notifications/cancelled", json!({"requestId": 1}));
+    rig.notify("notifications/cancelled", json!({"requestId": 99}));
+    rig.notify("notifications/cancelled", json!({"requestId": [1], "reason": "junk"}));
+    rig.notify("notifications/cancelled", json!("junk"));
+    rig.run(2, "quick");
+    assert!(rig.reply(2, QUICK).unwrap().get("result").is_some());
+    rig.finish();
+}
+
+#[test]
+fn initialize_cannot_be_cancelled() {
+    let mut rig = Rig::new(LONG, Duration::from_secs(60));
+    rig.request(1, "initialize", json!({}));
+    rig.notify("notifications/cancelled", json!({"requestId": 1}));
+    assert!(rig.reply(1, QUICK).unwrap().get("result").is_some());
+    rig.finish();
+}
+
+#[test]
+fn progress_reaches_a_client_that_gave_a_token_before_the_reply() {
+    let mut rig = Rig::new(LONG, Duration::from_secs(60));
+    rig.request(1, "tools/call", json!({"name": "command_run", "arguments": {"id": "progress.demo"}, "_meta": {"progressToken": "tok"}}));
+    let mut seen = vec![];
+    let reply = loop {
+        let v = rig.next(QUICK).expect("a reply");
+        if v["method"] == "notifications/progress" {
+            seen.push(v["params"].clone());
+        } else {
+            break v;
+        }
+    };
+    assert!(reply.get("result").is_some(), "{reply}");
+    assert_eq!(seen.len(), 3, "{seen:?}");
+    let values: Vec<u64> = seen.iter().map(|p| p["progress"].as_u64().unwrap()).collect();
+    assert_eq!(values, [1, 2, 3]);
+    assert!(seen.iter().all(|p| p["progressToken"] == "tok" && p["total"] == 3), "{seen:?}");
+    assert_eq!(seen[1]["message"], "page 2");
+    // Without a token nothing is sent.
+    rig.run(2, "progress.demo");
+    let next = rig.next(QUICK).unwrap();
+    assert_eq!(next["id"], 2, "{next}");
+    rig.finish();
+}
+
+/// The real engine behind the server: waits (like a slow render would) until the call is
+/// cancelled, then lets the export run into the cancelled token.
+struct Gated {
+    inner: crate::Headless,
+    started: Sender<()>,
+}
+
+impl Backend for Gated {
+    fn call(&mut self, method: &str, params: Value) -> Result<Value, String> {
+        if method == "engine.execute" && params["command"] == "document.export" {
+            self.started.send(()).ok();
+            let end = Instant::now() + LONG;
+            while !cancel::is_cancelled() && Instant::now() < end {
+                std::thread::sleep(Duration::from_millis(5));
+            }
+        }
+        self.inner.call(method, params)
+    }
+
+    fn has_ui(&self) -> bool {
+        false
+    }
+
+    fn describe(&self) -> String {
+        "gated".into()
+    }
+}
+
+#[test]
+fn a_cancelled_export_through_the_real_engine_writes_nothing_and_leaves_the_session_usable() {
+    let path = vectorcraft_testkit::temp_dir("mcp-cancel").join("out.pdf");
+    std::fs::remove_file(&path).ok();
+    let (read, mut to_server) = std::io::pipe().unwrap();
+    let (tx, replies) = mpsc::channel();
+    let (started_tx, started) = mpsc::channel();
+    let thread = std::thread::spawn(move || {
+        let backend = Gated { inner: crate::Headless::with_document(), started: started_tx };
+        Server::new(Box::new(backend)).serve(BufReader::new(read), Lines(tx, vec![])).unwrap();
+    });
+    let send = |w: &mut std::io::PipeWriter, v: Value| writeln!(w, "{v}").unwrap();
+    let params = json!({"name": "command_run", "arguments": {"id": "document.export", "params": {"path": path.to_str().unwrap()}}});
+    send(&mut to_server, json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": params}));
+    started.recv_timeout(QUICK).unwrap();
+    send(&mut to_server, json!({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 1}}));
+    // The next request is answered: the cancelled export neither blocks nor breaks the engine.
+    send(&mut to_server, json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "doc_inspect", "arguments": {}}}));
+    let reply: Value = serde_json::from_str(&replies.recv_timeout(QUICK).unwrap()).unwrap();
+    assert_eq!(reply["id"], 2, "request 1 is not answered: {reply}");
+    assert_eq!(reply["result"]["isError"], false, "{reply}");
+    assert!(!path.exists(), "a cancelled export writes no file");
+    drop(to_server);
+    thread.join().unwrap();
+}
+
+#[test]
+fn a_remote_call_stops_waiting_for_the_app_when_cancelled() {
+    use std::io::BufRead;
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    // An app that takes the request and never answers.
+    let app = std::thread::spawn(move || {
+        let (conn, _) = listener.accept().unwrap();
+        let mut line = String::new();
+        BufReader::new(conn.try_clone().unwrap()).read_line(&mut line).unwrap();
+        // Hold the connection open until the test is done with it.
+        std::thread::sleep(Duration::from_secs(3));
+        drop(conn);
+    });
+    let mut remote = crate::Remote::connect(&addr).unwrap();
+    let token = cancel::CancelToken::new();
+    let raise = token.clone();
+    std::thread::spawn(move || {
+        std::thread::sleep(Duration::from_millis(200));
+        raise.cancel();
+    });
+    let start = Instant::now();
+    let r = cancel::scope(&token, None, || remote.call("engine.execute", json!({"command": "document.export"})));
+    let why = r.unwrap_err();
+    assert!(why.contains("cancelled"), "{why}");
+    assert!(start.elapsed() < Duration::from_secs(2), "returned when cancelled, not when the app answered");
+    app.join().unwrap();
+}

crates/mcp/tests/protocol_props.rs

@@ -617,3 +617,64 @@ fn tools_never_leave_interactions_open() {
     }
     assert!(failures.is_empty(), "{}", failures.join("\n"));
 }
+
+proptest! {
+    #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })]
+
+    /// Requests mixed with `notifications/cancelled` (for ids that are queued, running, finished,
+    /// unknown or junk) read from one stream: the server ends, every request nobody cancelled is
+    /// answered exactly once, a cancelled one at most once, nothing is answered that wasn't asked,
+    /// replies keep the order of the requests, and the document stays valid.
+    #[test]
+    fn cancellation_never_loses_or_duplicates_an_answer(
+        steps in prop::collection::vec((0u8..6, 0u64..10), 1..40),
+    ) {
+        let mut input = String::new();
+        let mut asked: Vec<u64> = vec![];
+        let mut cancelled: Vec<u64> = vec![];
+        for (i, &(kind, n)) in steps.iter().enumerate() {
+            let id = i as u64 + 1;
+            let line = match kind {
+                0 => json!({"jsonrpc": "2.0", "id": id, "method": "ping"}),
+                1 => json!({"jsonrpc": "2.0", "id": id, "method": "tools/list"}),
+                2 => json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": {"name": "draw_shape", "arguments": {"shape": "rectangle", "x": 1, "y": 2, "width": 30, "height": 20}}}),
+                3 => json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": {"name": "command_run", "arguments": {"id": "select.all"}}, "_meta": {"progressToken": id}}),
+                // A cancel for an id that may be queued, finished or never asked.
+                4 => {
+                    let target = n.min(id.saturating_sub(1)).max(1);
+                    cancelled.push(target);
+                    input.push_str(&format!("{}\n", json!({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": target}})));
+                    continue;
+                }
+                // Junk cancels never take anything with them.
+                _ => {
+                    let junk = [json!(null), json!([1]), json!({"a": 1}), json!("nope"), json!(1.5), json!(-1)];
+                    input.push_str(&format!("{}\n", json!({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": junk[n as usize % junk.len()]}})));
+                    continue;
+                }
+            };
+            asked.push(id);
+            input.push_str(&format!("{line}\n"));
+        }
+        let mut s = server();
+        let mut out = Vec::new();
+        s.serve(input.as_bytes(), &mut out).unwrap();
+        let ids: Vec<u64> = String::from_utf8(out)
+            .unwrap()
+            .lines()
+            .map(|l| serde_json::from_str::<Value>(l).unwrap())
+            .filter(|v| v.get("method").is_none())
+            .map(|v| v["id"].as_u64().unwrap())
+            .collect();
+        prop_assert!(ids.windows(2).all(|w| w[0] < w[1]), "answered once each, in request order: {:?}", ids);
+        for id in &asked {
+            if !cancelled.contains(id) {
+                prop_assert!(ids.contains(id), "request {} was never answered: {:?}", id, ids);
+            }
+        }
+        for id in &ids {
+            prop_assert!(asked.contains(id), "answer to a request nobody made: {}", id);
+        }
+        check_document(&document(&mut s)).unwrap();
+    }
+}

crates/pdf/src/export.rs

@@ -44,8 +44,11 @@ pub fn export_with_report(doc: &Document, opts: &PdfOptions) -> Result<ExportRep
 fn export_with_report_inner(doc: &Document, opts: &PdfOptions) -> Result<ExportReport, PdfError> {
     let set = &opts.settings;
     set.check()?;
+    vectorcraft_doc::cancel::check().map_err(|_| PdfError::Cancelled)?;
     // Live geometry effects export as their result; raster effects are reported below.
     let baked = vectorcraft_effects::bake_document(doc);
+    // Baking stops early when cancelled, leaving a half-baked copy that must not be written.
+    vectorcraft_doc::cancel::check().map_err(|_| PdfError::Cancelled)?;
     let doc = baked.as_ref().unwrap_or(doc);
     let pages = pages(doc, opts)?;
 
@@ -77,7 +80,11 @@ fn export_with_report_inner(doc: &Document, opts: &PdfOptions) -> Result<ExportR
     // [`MAX_PAGE_SIDE`] is written in larger units; other files have every page in points.
     let scales = set.compatibility.has_user_unit() && !set.standard.is_pdfx();
     let mut units = Vec::with_capacity(pages.len());
-    for (i, boxes) in pages {
+    let total = pages.len() as u64;
+    for (written, (i, boxes)) in pages.into_iter().enumerate() {
+        // The page count and the page being written, for a client that asked for progress.
+        vectorcraft_doc::cancel::check().map_err(|_| PdfError::Cancelled)?;
+        vectorcraft_doc::cancel::progress(written as u64 + 1, total, &format!("Writing page {} of {total}", written + 1));
         let (m, r) = (boxes.media, boxes.bleed);
         let info = if marks.page_info { crate::marks::page_info(doc, &title, i, created_at) } else { String::new() };
         let art = crate::marks::art(doc, &marks, &boxes, bleed, &info);
@@ -103,6 +110,7 @@ fn export_with_report_inner(doc: &Document, opts: &PdfOptions) -> Result<ExportR
         };
         w.page(&mut ex, &sheet).map_err(|_| PdfError::BadArtboard(i))?;
     }
+    vectorcraft_doc::cancel::check().map_err(|_| PdfError::Cancelled)?;
     let (layers, overprinted) = (ex.layers, ex.overprinted);
     w.absorb(ex);
     let (bytes, more) = w.finish()?;

crates/pdf/src/lib.rs

@@ -205,6 +205,9 @@ pub enum PdfError {
     Unsupported(String),
     #[error("PDF writer error: {0}")]
     Write(String),
+    /// The caller asked for the export to stop ([`vectorcraft_doc::cancel`]).
+    #[error("cancelled")]
+    Cancelled,
     #[error("cannot read PDF: {0}")]
     Parse(String),
     #[error("the PDF has no pages")]

crates/render/src/lib.rs

@@ -1131,6 +1131,10 @@ impl Renderer {
         if !force && (!n.visible || f.opts.hidden.contains(&n.id)) {
             return;
         }
+        // A cancelled export stops drawing; whoever asked for the picture drops it (see `cancel`).
+        if vectorcraft_doc::cancel::is_cancelled() {
+            return;
+        }
         if force {
             // Bounds were already checked by draw_arc (or the caller wants it drawn regardless).
         } else if let Some(b) = fx::cull_bounds(f.symbols, n) {

docs/mcp.md

@@ -225,36 +225,57 @@ crate that installed one would override whatever logger an embedder had already
 `vectorcraft-mcp` directly means calling `vectorcraft_mcp::logging::install()` yourself if you want
 log records to reach your client.
 
-### Synchronous calls
+### Long calls: cancellation, progress and a busy engine
 
-Exports complete synchronously. A `_meta.progressToken` is harmlessly ignored, as are
-`notifications/cancelled` (including unknown request ids). There are no background export jobs
-or progress notifications; cancellation cannot interrupt a running call.
+A call runs to the end before its reply, and the engine runs one call at a time, in the order the
+lines arrived. A thread of its own reads the stream meanwhile, so the server still hears the client
+while an export runs (#2165, #2171, #2172):
+
+- **Cancellation.** `notifications/cancelled {requestId}`
+  ([spec](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation))
+  takes a request out of the queue, or stops the one that is running, and **that request gets no
+  reply**. A running export, flatten, bake or render notices at its next checkpoint (between pages
+  and artboards, between flattened groups, while drawing objects), returns, and writes nothing: the
+  document is exactly as it was and the engine is free for the next call. An unknown, finished or
+  malformed `requestId` is ignored, and so is a cancel of `initialize`. A call without checkpoints
+  (every ordinary edit is over in milliseconds) runs to its end, but its reply is dropped all the
+  same. Against a running app (`--connect`) the server stops waiting and drops the connection, but
+  the app cannot be told to stop what it started.
+- **Progress.** A request with `_meta.progressToken` gets `notifications/progress`
+  ([spec](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress)) before
+  its reply: the PDF writer reports "Writing page 2 of 5" with `progress` 2 and `total` 5, raster
+  export reports each artboard, and the PDF flattening stage reports a message without a total. `progress` only grows, updates come at most every 200 ms (the last page always),
+  and none follows the reply. A client that does not read fast enough misses updates, never the reply.
+- **A busy engine.** `ping`, `tools/list`, `resources/list`, `resources/templates/list`,
+  `prompts/list` and `prompts/get` need no engine and are answered at once, whatever is running.
+  Anything else that arrives while a call has already run for two seconds is not queued behind it:
+  it is answered with error `-32003` (`data.busy` names the running call, its request id and how
+  long it has run) and was **not run**, so the client can retry later or cancel the running call.
+  Requests that arrive sooner than that queue and are answered in order, as before, so a client
+  that sends a few quick calls together sees no difference. At most 64 lines wait; past that the
+  server stops reading until the engine catches up.
+
+`vectorcraft_mcp::Server::serve` takes the input and the output on separate threads (`Send`), and
+`Server::with_busy_after` changes the two seconds. Calling `handle_line` directly is as before:
+one line in, one reply out, no threads.
 
 ### Not implemented
 
-Everything below needs the same thing first: the server reads one line at a time and answers it with
-one reply, so **while a `tools/call` is running it neither reads an incoming line nor writes an
-outgoing one.** Progress and cancellation are therefore both meaningless as they stand, and a
-subscription could only ever be checked between calls — which is not what a client asking to be told
-about a change expects.
-
 - **`resources/subscribe` / `notifications/resources/updated`.** It would also have to be honest about
   *whose* edits it sees: comparing the undo history's length misses edits made in the app window,
   which is the main reason to run against a live document. `document.inspect`'s `revision` is the
   right signal — it is monotonic and already bumped when a document is replaced — but noticing an
   edit still means someone asking.
-- **`notifications/progress`.** It would have to be written *before* the reply it belongs to, since a
-  progress token stops meaning anything once the request has completed.
-- **`notifications/cancelled`.** It would have to be read *during* the call it cancels.
-- **`sampling/createMessage` and `elicitation/create`.** Both are server→client requests, blocked by
-  the same one-line-one-reply loop. Supporting them means being able to suspend a tool call
-  mid-flight, which is what the `2026-07-28` revision's input-requests mechanism exists for.
+- **Reading the document during a long call.** Requests that need the engine (`doc_inspect`,
+  `resources/read`, `command_list`) are refused as busy while an export runs, not answered from a
+  snapshot.
+- **`sampling/createMessage` and `elicitation/create`.** Both are server→client requests. Supporting
+  them means being able to suspend a tool call mid-flight, which is what the `2026-07-28` revision's
+  input-requests mechanism exists for.
 - **Streamable HTTP / SSE.** stdio only.
 
 `completion/complete` covers the "what could I pass here" half of the same problem, which is why it
-is here and subscriptions are not. The fix for the rest is a reader thread (or an event loop) around
-the transport — a design change rather than a feature, so it is not in this crate yet.
+is here and subscriptions are not.
 
 ## Tools