MCP: report render progress and handle cancellation

Keep serving during long renders, recover failed workers, and remove only files opened by the interrupted export. Cover progress, cancellation, EOF and output ownership with regression tests.

Co-Authored-By: OpenAI Codex <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/automation/src/lib.rs

@@ -19,6 +19,7 @@ mod autosave;
 pub mod backend;
 pub mod base64;
 pub mod bridge;
+mod long_job;
 pub mod server;
 pub mod tools;
 

crates/automation/src/long_job.rs

@@ -0,0 +1,177 @@
+//! Long renders over MCP (docs/mcp.md "Progress and cancellation"). While [`McpServer::serve`]
+//! runs, a blocking `renderQueue.render` (`command_run`, headless) renders on the session's
+//! background render job instead of inline: the server keeps reading stdin, answers every other
+//! request in between (the job renders a snapshot of the project), reports progress as
+//! `notifications/progress` when the request carried `_meta.progressToken`, and stops the render
+//! on `notifications/cancelled` (deleting the partial output, sending no response).
+
+use std::io::Write;
+use std::sync::mpsc::{Receiver, RecvTimeoutError};
+use std::time::Duration;
+
+use serde_json::{Value, json};
+
+use crate::server::{McpServer, call_result};
+use crate::tools::{self, Reply};
+
+/// A `tools/call` result for `call`, with `resultType` for modern clients.
+fn result_for(call: &LongCall, r: Result<Reply, crate::Error>) -> Value {
+    let mut result = call_result(r);
+    if call.modern {
+        result["resultType"] = json!("complete");
+    }
+    result
+}
+
+/// How often the job is polled (and at most how often progress is reported).
+const POLL: Duration = Duration::from_millis(100);
+
+/// A `tools/call` that runs as a long job.
+pub struct LongCall {
+    pub id: Value,
+    pub params: Value,
+    pub token: Option<Value>,
+    /// The request declared MCP 2026-07-28+ (its result then carries `resultType`).
+    pub modern: bool,
+}
+
+/// Whether `msg` is a blocking `renderQueue.render` call (valid arguments, headless backend).
+pub fn long_call(server: &mut McpServer, msg: &Value) -> Option<LongCall> {
+    let id = msg.get("id").filter(|i| i.is_string() || i.is_number())?;
+    if msg.get("method").and_then(Value::as_str) != Some("tools/call") || server.backend().is_bridge() {
+        return None;
+    }
+    let p = msg.get("params")?;
+    let name = p.get("name")?.as_str()?;
+    let args = p.get("arguments").cloned().unwrap_or(json!({}));
+    if !matches!(name, "command_run" | "execute_command")
+        || args.get("id").or_else(|| args.get("command")).and_then(Value::as_str) != Some("renderQueue.render")
+    {
+        return None;
+    }
+    if tools::find(name).and_then(|t| t.unknown_arg(&args)).is_some() {
+        return None; // the normal path reports it
+    }
+    let params = match args.get("params") {
+        None | Some(Value::Null) => json!({}),
+        Some(v) if v.is_object() => v.clone(),
+        _ => return None,
+    };
+    if params.get("wait").and_then(Value::as_bool) == Some(false) {
+        return None;
+    }
+    let token = p.get("_meta").and_then(|m| m.get("progressToken")).filter(|t| t.is_string() || t.is_number()).cloned();
+    Some(LongCall { id: id.clone(), params, token, modern: crate::server::is_modern(p) })
+}
+
+fn send(out: &mut impl Write, v: &Value) -> std::io::Result<()> {
+    out.write_all(v.to_string().as_bytes())?;
+    out.write_all(b"\n")?;
+    out.flush()
+}
+
+/// The `requestId` of a `notifications/cancelled` line.
+fn cancelled_id(line: &str) -> Option<Value> {
+    let v: Value = serde_json::from_str(line).ok()?;
+    (v.get("method")?.as_str()? == "notifications/cancelled").then(|| v.get("params")?.get("requestId").cloned())?
+}
+
+impl McpServer {
+    /// Run `call` to completion or cancellation, handling the other messages from `inbox`
+    /// (`None` = stdin closed) meanwhile. Returns false when stdin closed.
+    pub(crate) fn run_long(&mut self, call: LongCall, inbox: &Receiver<Option<String>>, out: &mut impl Write) -> std::io::Result<bool> {
+        let mut params = call.params.clone();
+        params["wait"] = json!(false);
+        let start = self.guarded("command_run", &json!({"id": "renderQueue.render", "params": params}));
+        let items = match &start {
+            Ok(Reply::Json(v)) => v["items"].as_array().cloned().unwrap_or_default(),
+            _ => {
+                send(out, &json!({"jsonrpc": "2.0", "id": call.id, "result": result_for(&call, start)}))?;
+                return Ok(true);
+            }
+        };
+        let ids: Vec<u64> = items.iter().filter_map(|i| i["id"].as_u64()).collect();
+        let shared = self.backend().session().and_then(|s| s.render_job.as_ref().map(|job| job.shared.clone()));
+        let mut last = -1.0f64;
+        let mut last_report: Option<std::time::Instant> = None;
+        let mut open = true;
+        let mut cancelled = false;
+        loop {
+            let state = shared.as_ref().map(|job| job.snapshot());
+            let rendering = state.as_ref().is_some_and(|state| !state.finished);
+            if let (Some(token), Some(st)) = (&call.token, &state)
+                && st.items_total > 0
+                && !cancelled
+            {
+                let frac = if st.total > 0 { st.done as f64 / st.total as f64 } else { 0.0 };
+                // the current item's frames count until it is done
+                let progress = (st.items_done as f64 + if st.items_done < st.items_total { frac.min(1.0) } else { 0.0 }).min(st.items_total as f64);
+                if progress > last && (!rendering || last_report.is_none_or(|at| at.elapsed() >= POLL)) {
+                    last = progress;
+                    last_report = Some(std::time::Instant::now());
+                    let message = format!(
+                        "Rendering item {} of {}: frame {} of {}",
+                        st.items_done.saturating_add(1).min(st.items_total),
+                        st.items_total,
+                        st.done,
+                        st.total
+                    );
+                    send(
+                        out,
+                        &json!({"jsonrpc": "2.0", "method": "notifications/progress",
+                        "params": {"progressToken": token, "progress": progress, "total": st.items_total, "message": message}}),
+                    )?;
+                }
+            }
+            if !rendering {
+                break;
+            }
+            if !open {
+                // stdin closed: finish the render (a piped script expects its reply), read nothing more
+                std::thread::sleep(POLL);
+                continue;
+            }
+            match inbox.recv_timeout(POLL) {
+                Ok(Some(line)) => {
+                    if cancelled_id(&line).as_ref() == Some(&call.id) {
+                        cancelled = true;
+                        if let Some(job) = &shared {
+                            job.cancel.store(true, std::sync::atomic::Ordering::Relaxed);
+                        }
+                        continue;
+                    }
+                    if let Some(other) = serde_json::from_str::<Value>(&line).ok().and_then(|msg| long_call(self, &msg)) {
+                        let result = result_for(&other, Err(crate::Error::Other("a render is already in progress".into())));
+                        send(out, &json!({"jsonrpc":"2.0", "id":other.id, "result":result}))?;
+                        continue;
+                    }
+                    if let Some(reply) = self.handle_line(&line) {
+                        out.write_all(reply.as_bytes())?;
+                        out.write_all(b"\n")?;
+                        out.flush()?;
+                    }
+                }
+                Ok(None) | Err(RecvTimeoutError::Disconnected) => open = false,
+                Err(RecvTimeoutError::Timeout) => {}
+            }
+        }
+        let Some(s) = self.backend().session() else { return Ok(open) };
+        s.poll_render();
+        s.drain_events();
+        let list = self.guarded("command_run", &json!({"id": "renderQueue.list"}));
+        let items: Vec<Value> = match &list {
+            Ok(Reply::Json(v)) => {
+                v["items"].as_array().cloned().unwrap_or_default().into_iter().filter(|i| i["id"].as_u64().is_some_and(|id| ids.contains(&id))).collect()
+            }
+            _ => Vec::new(),
+        };
+        if cancelled {
+            // Exporters remove only files they actually opened for this interrupted output.
+            return Ok(open); // MCP: no response for a cancelled request
+        }
+        let failed = ids.len().saturating_sub(items.len()).saturating_add(items.iter().filter(|item| item["statusLabel"].as_str() != Some("Done")).count());
+        let result = result_for(&call, Ok(Reply::Json(json!({"rendering": false, "items": items, "failed": failed}))));
+        send(out, &json!({"jsonrpc": "2.0", "id": call.id, "result": result}))?;
+        Ok(open)
+    }
+}

crates/automation/src/server.rs

@@ -1,5 +1,5 @@
 //! MCP server: JSON-RPC 2.0, one message per line (the MCP stdio transport). Implements
-//! `initialize`, `ping`, `tools/list` and `tools/call`; notifications are accepted and ignored.
+//! tools and JSON resources. The stdio loop handles progress and cancellation for long renders.
 
 use std::io::{BufRead, Write};
 use std::path::Path;
@@ -101,18 +101,48 @@ impl McpServer {
         Some(reply.to_string())
     }
 
-    /// Serve until EOF on `input`.
-    pub fn serve(&mut self, input: impl BufRead, mut output: impl Write) -> std::io::Result<()> {
-        let served = (|| {
-            for line in input.lines() {
-                if let Some(reply) = self.handle_line(&line?) {
+    /// Serve until EOF on `input`. Input is read on its own thread so a long render can report
+    /// progress, be cancelled and let other requests through while it runs (`long_job`).
+    pub fn serve(&mut self, input: impl BufRead + Send, mut output: impl Write) -> std::io::Result<()> {
+        let (tx, inbox) = std::sync::mpsc::channel::<Option<String>>();
+        // A read error ends the input like EOF and is returned once the queued requests are answered.
+        let read_error = std::sync::Mutex::new(None::<std::io::Error>);
+        let served = std::thread::scope(|scope| {
+            let read_error = &read_error;
+            std::thread::Builder::new().name("mcp-input".into()).spawn_scoped(scope, move || {
+                for line in input.lines() {
+                    let line = match line {
+                        Ok(line) => line,
+                        Err(e) => {
+                            *read_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(e);
+                            break;
+                        }
+                    };
+                    if tx.send(Some(line)).is_err() {
+                        return;
+                    }
+                }
+                let _ = tx.send(None);
+            })?;
+            while let Ok(Some(line)) = inbox.recv() {
+                let long = serde_json::from_str::<Value>(line.trim()).ok().and_then(|m| crate::long_job::long_call(self, &m));
+                if let Some(call) = long {
+                    if !self.run_long(call, &inbox, &mut output)? {
+                        break;
+                    }
+                    continue;
+                }
+                if let Some(reply) = self.handle_line(&line) {
                     output.write_all(reply.as_bytes())?;
                     output.write_all(b"\n")?;
                     output.flush()?;
                 }
             }
-            Ok(())
-        })();
+            match read_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take() {
+                Some(e) => Err(e),
+                None => Ok(()),
+            }
+        });
         match (served, self.checkpoint(true)) {
             (r, Ok(())) => r,
             (Ok(()), Err(e)) => Err(e),
@@ -122,7 +152,7 @@ impl McpServer {
 
     /// Serve on stdin/stdout (the MCP stdio transport).
     pub fn serve_stdio(&mut self) -> std::io::Result<()> {
-        self.serve(std::io::stdin().lock(), std::io::stdout().lock())
+        self.serve(std::io::BufReader::new(std::io::stdin()), std::io::stdout().lock())
     }
 
     fn request(&mut self, method: &str, params: &Value) -> Result<Value, (i64, String)> {

crates/automation/src/tests.rs

@@ -874,3 +874,182 @@ fn render_job_state_recovers_a_poisoned_lock() {
     .join();
     assert_eq!(shared.snapshot().done, 3);
 }
+
+/// A render that takes ~2 ms per frame and writes its file first (so a cancel leaves a partial
+/// one to clean up).
+struct SlowExporter;
+
+impl effectcraft_engine::Exporter for SlowExporter {
+    fn formats(&self) -> Vec<effectcraft_engine::project::render_queue::OutputFormat> {
+        effectcraft_engine::project::render_queue::OutputFormat::ALL.to_vec()
+    }
+    fn export(&self, job: &effectcraft_engine::ExportJob, progress: &mut dyn FnMut(u64, u64) -> bool) -> Result<effectcraft_engine::ExportResult, String> {
+        std::fs::write(job.path, b"partial").map_err(|e| e.to_string())?;
+        let n = 2000;
+        for k in 0..=n {
+            if !progress(k, n) {
+                std::fs::remove_file(job.path).map_err(|e| e.to_string())?;
+                return Err(effectcraft_engine::render_queue::CANCELLED.into());
+            }
+            std::thread::sleep(std::time::Duration::from_millis(2));
+        }
+        Ok(effectcraft_engine::ExportResult { path: job.path.to_string(), frames: n, ..Default::default() })
+    }
+}
+
+/// docs/mcp.md "Progress and cancellation" over the stdio loop: a blocking `renderQueue.render`
+/// reports `notifications/progress` for its token, other requests are answered while it renders,
+/// and `notifications/cancelled` stops it, deletes the partial file and sends no response.
+#[test]
+fn render_progress_and_cancel() {
+    let dir = std::env::temp_dir().join(format!("effectcraft-mcp-progress-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let (in_r, mut in_w) = std::io::pipe().unwrap();
+    let (out_r, out_w) = std::io::pipe().unwrap();
+    let server = std::thread::spawn(move || {
+        let session = Session { exporter: Some(std::sync::Arc::new(SlowExporter)), ..Default::default() };
+        McpServer::new(Backend::headless(session)).serve(BufReader::new(in_r), out_w).unwrap();
+    });
+    let mut lines = BufReader::new(out_r).lines();
+    let mut next = || -> Value { serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap() };
+    let mut send = |v: Value| writeln!(in_w, "{v}").unwrap();
+    let run = |id: u64, cmd: &str, params: Value, token: Option<Value>| {
+        let mut p = json!({"name": "command_run", "arguments": {"id": cmd, "params": params}});
+        if let Some(t) = token {
+            p["_meta"] = json!({"progressToken": t, "io.modelcontextprotocol/protocolVersion":"2026-07-28"});
+        }
+        json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": p})
+    };
+    send(json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}));
+    next();
+    send(run(2, "comp.new", json!({"width": 64, "height": 64, "duration": 1}), None));
+    assert_eq!(next()["result"]["isError"], false);
+    let a = dir.join("a.mp4").to_string_lossy().to_string();
+    send(run(3, "renderQueue.add", json!({"output": a}), None));
+    assert_eq!(next()["result"]["isError"], false);
+    // progress for the token, strictly increasing; doc_inspect answered while rendering
+    send(run(4, "renderQueue.render", json!({}), Some(json!("tok"))));
+    let (mut last, mut notes, mut asked, mut answered) = (-1.0, 0, false, false);
+    let done = loop {
+        let m = next();
+        if m["method"] == "notifications/progress" {
+            assert_eq!(m["params"]["progressToken"], "tok", "{m}");
+            let p = m["params"]["progress"].as_f64().unwrap();
+            assert!(p > last && p <= m["params"]["total"].as_f64().unwrap(), "{m}");
+            (last, notes) = (p, notes + 1);
+            if !asked && p > 0.0 {
+                asked = true;
+                send(json!({"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "doc_inspect", "arguments": {}}}));
+            }
+        } else if m["id"] == 5 {
+            answered = true;
+            assert_eq!(m["result"]["isError"], false, "{m}");
+        } else {
+            assert_eq!(m["id"], 4, "{m}");
+            break m;
+        }
+    };
+    assert_eq!(done["result"]["isError"], false, "{done}");
+    assert_eq!(done["result"]["resultType"], "complete");
+    assert_eq!(last, 1.0, "final progress reaches the total");
+    let v: Value = serde_json::from_str(done["result"]["content"][0]["text"].as_str().unwrap()).unwrap();
+    assert_eq!(v["items"][0]["statusLabel"], "Done", "{v}");
+    assert!(notes >= 2 && answered, "{notes} notifications; doc_inspect answered during the render: {answered}");
+    // cancel after the first notification: no response, the partial file is gone
+    let b = dir.join("b.mp4");
+    send(run(6, "renderQueue.add", json!({"output": b.to_string_lossy()}), None));
+    assert_eq!(next()["result"]["isError"], false);
+    send(run(7, "renderQueue.render", json!({}), Some(json!(7))));
+    loop {
+        let m = next();
+        assert_ne!(m["id"], 7, "{m}");
+        if m["method"] == "notifications/progress" && m["params"]["progress"].as_f64().unwrap() > 0.0 {
+            break;
+        }
+    }
+    assert!(b.exists(), "the render started writing");
+    send(json!({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 7, "reason": "test"}}));
+    send(json!({"jsonrpc": "2.0", "id": 8, "method": "ping"}));
+    let m = loop {
+        let m = next();
+        if m["method"] != "notifications/progress" {
+            break m;
+        }
+    };
+    assert_eq!(m["id"], 8, "no response for the cancelled request: {m}");
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
+    let v = loop {
+        send(run(9, "renderQueue.list", json!({}), None));
+        let v: Value = serde_json::from_str(next()["result"]["content"][0]["text"].as_str().unwrap()).unwrap();
+        // the queue stops rendering a moment before the stopped item's status is updated
+        if v["rendering"] == false && v["items"][1]["statusLabel"] != "Rendering" {
+            break v;
+        }
+        assert!(std::time::Instant::now() < deadline, "cancel did not stop the job: {v}");
+        std::thread::sleep(std::time::Duration::from_millis(20));
+    };
+    assert!(!b.exists(), "partial output deleted");
+    assert!(std::path::Path::new(&a).exists(), "completed earlier queue item is retained");
+    assert_eq!(v["items"][1]["statusLabel"], "User Stopped", "{v}");
+    drop(send);
+    drop(in_w);
+    server.join().unwrap();
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn render_without_token_finishes_after_stdin_closes() {
+    let dir = std::env::temp_dir().join(format!("ec-mcp-eof-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let path = dir.join("complete.mp4");
+    let mut s = McpServer::new(Backend::headless(Session { exporter: Some(std::sync::Arc::new(SlowExporter)), ..Default::default() }));
+    let mut input = String::new();
+    for (id, command, params) in
+        [(1, "comp.new", json!({"width":16,"height":16})), (2, "renderQueue.add", json!({"output":path})), (3, "renderQueue.render", json!({}))]
+    {
+        input.push_str(&format!(
+            "{}\n",
+            json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"command_run","arguments":{"id":command,"params":params}}})
+        ));
+    }
+    let mut out = Vec::new();
+    s.serve(std::io::Cursor::new(input), &mut out).unwrap();
+    let replies: Vec<Value> = String::from_utf8(out).unwrap().lines().map(|line| serde_json::from_str(line).unwrap()).collect();
+    assert_eq!(replies.len(), 3, "no token means no notifications: {replies:?}");
+    assert_eq!(replies[2]["result"]["isError"], false);
+    assert!(path.exists());
+    std::fs::remove_dir_all(dir).unwrap();
+}
+
+#[test]
+fn panicked_render_worker_finishes_with_an_error() {
+    struct BrokenExporter;
+    impl effectcraft_engine::Exporter for BrokenExporter {
+        fn formats(&self) -> Vec<effectcraft_engine::project::render_queue::OutputFormat> {
+            effectcraft_engine::project::render_queue::OutputFormat::ALL.to_vec()
+        }
+        fn export(&self, _: &effectcraft_engine::ExportJob, _: &mut dyn FnMut(u64, u64) -> bool) -> Result<effectcraft_engine::ExportResult, String> {
+            panic!("synthetic exporter failure");
+        }
+    }
+    let mut session = Session { exporter: Some(std::sync::Arc::new(BrokenExporter)), ..Default::default() };
+    session.execute("comp.new", json!({"width":16,"height":16})).unwrap();
+    session.execute("renderQueue.add", json!({"output":"unused.mp4"})).unwrap();
+    session.execute("renderQueue.render", json!({"wait":false})).unwrap();
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
+    while session.is_rendering() && std::time::Instant::now() < deadline {
+        std::thread::sleep(std::time::Duration::from_millis(10));
+    }
+    assert!(!session.is_rendering(), "a panicked worker must not leave an unfinishable job");
+    session.poll_render();
+    let queue = session.execute("renderQueue.list", json!({})).unwrap();
+    assert!(queue.to_string().contains("synthetic exporter failure"), "{queue}");
+    session.execute("renderQueue.add", json!({"output":"unused-again.mp4"})).unwrap();
+    let mut server = McpServer::new(Backend::headless(session));
+    let input =
+        json!({"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"command_run","arguments":{"id":"renderQueue.render"}}}).to_string() + "\n";
+    let mut out = Vec::new();
+    server.serve(std::io::Cursor::new(input), &mut out).unwrap();
+    let reply: Value = serde_json::from_slice(&out).unwrap();
+    assert_eq!(reply["result"]["isError"], true, "{reply}");
+}

crates/engine/src/render_queue.rs

@@ -58,7 +58,8 @@ pub trait Exporter: Send + Sync {
     /// Formats this build can write.
     fn formats(&self) -> Vec<OutputFormat>;
     /// Render and write one item (blocking). `progress(done, total)` returns `false` to cancel;
-    /// a cancelled export returns `Err("cancelled")`.
+    /// a cancelled export removes the files it opened for that output and returns `Err("cancelled")`.
+    /// Completed earlier queue items and unrelated files must be preserved.
     fn export(&self, job: &ExportJob, progress: &mut dyn FnMut(u64, u64) -> bool) -> Result<ExportResult, String>;
     /// [`Exporter::export`] as a future: renders whose GPU readbacks are deferred (a browser
     /// job worker's WebGPU device) await the device between a frame's passes. The default runs
@@ -494,7 +495,22 @@ impl Session {
             let sh = shared.clone();
             let thread = std::thread::Builder::new()
                 .name("render-queue".into())
-                .spawn(move || effectcraft_render::passes::block_on(run_work(work, &sh, &mut |_| {})))
+                .spawn(move || {
+                    let pending: Vec<u64> = work.items.iter().filter_map(|modules| modules.first().map(|(item, _)| item.id)).collect();
+                    if let Err(error) = crate::guard::guarded("render worker", || {
+                        effectcraft_render::passes::block_on(run_work(work, &sh, &mut |_| {}));
+                        Ok(())
+                    }) {
+                        let completed = lock(&sh.state).items_done;
+                        let mut updates = lock(&sh.updates);
+                        for id in pending.into_iter().skip(completed) {
+                            if !updates.iter().any(|u| matches!(u, ItemUpdate::Finished { id: done, .. } if *done == id)) {
+                                updates.push(ItemUpdate::Finished { id, status: RenderStatus::Failed(error.to_string()), seconds: 0.0, output: None });
+                            }
+                        }
+                        lock(&sh.state).finished = true;
+                    }
+                })
                 .map_err(|e| e.to_string())?;
             self.render_job = Some(RenderJob { shared, thread: Some(thread), remote: None });
         }

crates/export/src/encode.rs

@@ -277,7 +277,7 @@ pub(crate) async fn movie(job: &Cx<'_>, comp: &Comp, w: u32, h: u32, st: &mut St
     st.advance(first_end)?;
 
     let path = job.place(job.path, 1);
-    let mut file = crate::out::create(job.sink, &path)?;
+    let mut file = job.create(&path)?;
     let opts = WriterOptions::new(brand);
     let movie_ts = opts.movie_timescale.max(1) as i128;
     let mut mux: Writer = Mp4Writer::new(&mut file, opts).map_err(mux_err)?;

crates/export/src/lib.rs

@@ -171,6 +171,10 @@ pub async fn export_async(job: &Job<'_>, progress: &mut dyn FnMut(&Progress) ->
     let t0 = Instant::now();
     let cx = Cx::new(job);
     let mut r = export_with(&cx, progress, t0).await;
+    if matches!(r, Err(ExportError::Cancelled)) {
+        cx.remove_cancelled_outputs()?;
+        return r;
+    }
     let log = cx.write_log(&r, t0.elapsed().as_secs_f64());
     if let Ok(rep) = &mut r {
         rep.log = log;
@@ -263,7 +267,7 @@ async fn sequence(cx: &Cx<'_>, comp: &Comp, w: u32, h: u32, st: &mut State<'_>)
                 write_still(cx, &mut buf, fmt, &img, comp, channels, w, h, cx.output.quality)?;
                 let data = buf.into_inner();
                 let path = cx.place(&path, data.len() as u64);
-                let mut f = out::create(cx.sink, &path)?;
+                let mut f = cx.create(&path)?;
                 std::io::Write::write_all(&mut f, &data).map_err(io)?;
                 f.finish()
             })
@@ -350,7 +354,7 @@ async fn gif_export(job: &Cx<'_>, comp: &Comp, w: u32, h: u32, st: &mut State<'_
         return Err(ExportError::Unsupported("GIF frames are limited to 65535 px".into()));
     }
     let path = job.place(job.path, 1);
-    let mut file = out::create(job.sink, &path)?;
+    let mut file = job.create(&path)?;
     let mut enc = gif::Encoder::new(&mut file, w as u16, h as u16, &[]).map_err(|e| ExportError::Encode(e.to_string()))?;
     enc.set_repeat(if job.output.gif_loop { gif::Repeat::Infinite } else { gif::Repeat::Finite(0) }).map_err(|e| ExportError::Encode(e.to_string()))?;
     let rate = job.settings.rate(comp);

crates/export/src/pipeline.rs

@@ -24,6 +24,7 @@ pub(crate) struct Cx<'a> {
     pub frame_times: Option<Mutex<Vec<(u64, f64)>>>,
     /// Files written to an overflow folder.
     pub overflowed: Mutex<Vec<String>>,
+    written: Mutex<Vec<String>>,
 }
 
 impl<'a> std::ops::Deref for Cx<'a> {
@@ -68,7 +69,7 @@ impl<'a> Cx<'a> {
         let mut project = job.project.clone();
         apply_overrides(&mut project, job);
         let frame_times = (job.options.log == RenderLog::PlusPerFrameInfo).then(|| Mutex::new(Vec::new()));
-        Cx { job, project, frame_times, overflowed: Mutex::new(vec![]) }
+        Cx { job, project, frame_times, overflowed: Mutex::new(vec![]), written: Mutex::new(vec![]) }
     }
 
     pub fn comp(&self) -> Option<&Comp> {
@@ -187,6 +188,30 @@ impl<'a> Cx<'a> {
         crate::fit(px, img.width, img.height, w, h)
     }
 
+    /// Open a file and remember ownership for cancellation cleanup (including overflow paths).
+    pub fn create(&self, path: &str) -> crate::Result<out::Out<'_>> {
+        let file = out::create(self.sink, path)?;
+        if self.sink.is_none() {
+            self.written.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push(path.to_owned());
+        }
+        Ok(file)
+    }
+
+    pub fn remove_cancelled_outputs(&self) -> crate::Result<()> {
+        let mut failure = None;
+        for path in self.written.lock().unwrap_or_else(std::sync::PoisonError::into_inner).iter() {
+            if let Err(e) = std::fs::remove_file(path)
+                && e.kind() != std::io::ErrorKind::NotFound
+            {
+                failure.get_or_insert_with(|| crate::ExportError::Io(format!("cannot remove cancelled output {path}: {e}")));
+            }
+        }
+        match failure {
+            Some(error) => Err(error),
+            None => Ok(()),
+        }
+    }
+
     /// Where a file goes: `path`, or (Use Storage Overflow) the first overflow folder with room
     /// for `bytes` when the quota hook says the output volume is full.
     pub fn place(&self, path: &str, bytes: u64) -> String {
@@ -246,7 +271,7 @@ impl<'a> Cx<'a> {
                 s.push_str(&format!("  Frame {} ({:.3} s): rendered in {:.1} ms\n", f0 + i as i64, t.seconds(), secs * 1000.0));
             }
         }
-        let mut f = out::create(self.job.sink, &path).ok()?;
+        let mut f = self.create(&path).ok()?;
         std::io::Write::write_all(&mut f, s.as_bytes()).ok()?;
         f.finish().ok()?;
         Some(path)

crates/export/src/webm.rs

@@ -162,7 +162,7 @@ pub(crate) async fn webm(job: &Cx<'_>, comp: &Comp, w: u32, h: u32, st: &mut Sta
     let duration_ms = frame_ms(total) as f64;
 
     let path = job.place(job.path, 1);
-    let mut file = crate::out::create(job.sink, &path)?;
+    let mut file = job.create(&path)?;
     // EBML header.
     let mut head = vec![];
     let mut eb = vec![];
@@ -403,7 +403,7 @@ pub(crate) fn audio_file(job: &Cx, comp: &Comp, aiff: bool, st: &mut State) -> R
         }
     }
     let path = job.place(job.path, out.len() as u64);
-    let mut file = crate::out::create(job.sink, &path)?;
+    let mut file = job.create(&path)?;
     file.write_all(&out).map_err(io)?;
     let bytes = file.finish()?;
     Ok(Report { path, frames: 0, width: 0, height: 0, seconds: 0.0, bytes, audio: true, log: None, overflow: vec![] })

crates/export/src/webm_av1.rs

@@ -51,7 +51,7 @@ pub(crate) async fn webm_av1(job: &Cx<'_>, comp: &Comp, w: u32, h: u32, st: &mut
     let duration_ms = frame_ms(total) as f64;
 
     let path = job.place(job.path, 1);
-    let mut file = crate::out::create(job.sink, &path)?;
+    let mut file = job.create(&path)?;
     let mut head = vec![];
     let mut eb = vec![];
     el_uint(&mut eb, 0x4286, 1);

crates/export/tests/render_settings.rs

@@ -327,3 +327,61 @@ fn wav_and_aiff_channels_and_formats() {
     assert_eq!(&d[8..12], b"AIFF");
     assert_eq!(u16::from_be_bytes([d[26], d[27]]), 24, "sample size");
 }
+
+#[test]
+fn cancelled_export_removes_only_files_it_opened() {
+    use effectcraft_export::ExportError;
+    let dir = std::env::temp_dir().join(format!("ec-cancel-owned-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let (p, cid) = project(FrameRate::new(25, 1), |_, _| {});
+    struct Overflow;
+    impl StorageQuota for Overflow {
+        fn has_room(&self, path: &str, _: u64) -> bool {
+            path.contains("overflow")
+        }
+    }
+    for overflow in [false, true] {
+        let output_dir = if overflow { dir.join("overflow") } else { dir.clone() };
+        std::fs::create_dir_all(&output_dir).unwrap();
+        let settings = RenderSettings { skip_existing: true, storage_overflow: overflow, ..Default::default() };
+        for format in [OutputFormat::PngSequence, OutputFormat::H264] {
+            let om = OutputModule::for_format(format);
+            let path = dir.join(if format.is_sequence() { "clip_[#####].png" } else { "clip.mp4" });
+            let path = path.to_str().unwrap();
+            let decoys = ["clip.aep", "clip2.psd", "clip000.mp4", "clip_99999.png", "clip_00000.png", "clip.txt", "clip_RenderLog.txt"];
+            for name in decoys {
+                std::fs::write(output_dir.join(name), b"keep").unwrap();
+            }
+            // Skip-existing checks the primary destination even when the new frames overflow.
+            std::fs::write(dir.join("clip_00000.png"), b"keep").unwrap();
+            let result = export(
+                &Job {
+                    project: &p,
+                    footage: &NoFootage,
+                    expr: None,
+                    accel: None,
+                    comp: cid,
+                    settings: &settings,
+                    output: &om,
+                    path,
+                    sink: None,
+                    nested_switches: true,
+                    options: JobOptions {
+                        log: RenderLog::PlusPerFrameInfo,
+                        storage: Some(&Overflow),
+                        overflow: vec![output_dir.to_string_lossy().into_owned()],
+                        ..Default::default()
+                    },
+                },
+                &mut |progress| progress.done == 0,
+            );
+            assert!(matches!(result, Err(ExportError::Cancelled)), "{result:?}");
+            for name in decoys {
+                assert_eq!(std::fs::read(output_dir.join(name)).unwrap(), b"keep", "{name}");
+            }
+            let remaining: Vec<_> = std::fs::read_dir(&output_dir).unwrap().map(|e| e.unwrap().file_name().to_string_lossy().into_owned()).collect();
+            assert_eq!(remaining.len(), decoys.len(), "partial files remain: {remaining:?}");
+        }
+    }
+    std::fs::remove_dir_all(dir).unwrap();
+}

docs/agents.md

@@ -73,6 +73,8 @@ Without `--autosave`, headless work is in memory until explicitly saved. The fla
 only; a bridged desktop app owns its auto-saves. An edit interrupted before its reply is received
 may be absent from the last checkpoint.
 
+See [MCP conventions](mcp.md) for core tools, resources, strict arguments, and render progress/cancellation.
+
 ### Tools
 
 | Tool | What it does |

docs/mcp.md

@@ -34,3 +34,23 @@ receive `resultType: complete` and list/read cache hints; document reads are nev
 {"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"command_run","arguments":{"id":"comp.new","params":{"width":640,"height":360}}}}
 {"jsonrpc":"2.0","id":2,"method":"resources/read","params":{"uri":"effectcraft://document"}}
 ```
+
+## Progress and cancellation
+
+Headless `command_run` / `execute_command` calls to `renderQueue.render` (unless `wait: false`)
+run the existing background job while the MCP server keeps reading stdin. A
+`params._meta.progressToken` string or number opts into `notifications/progress`, at most ten
+per second, strictly increasing with a queue-item total. No token means no notifications.
+`ping`, inspection and other requests are served while rendering. The existing engine refuses
+a second render while one is active. Closing stdin lets the pending render finish and reply.
+
+Send `notifications/cancelled` with `params.requestId` to stop the matching request at the next
+frame batch. No response is sent for that cancelled request. Unknown and completed ids are
+ignored. The exporter records the exact files it opens, including storage overflow paths, and
+removes those files on cancellation; it never scans for similarly named files. Finished earlier
+queue items and skipped existing frames remain. Custom exporters implement the same cleanup
+contract. A failed queue item gives `isError: true`.
+
+Bridge calls, `wait: false`, batch tools and single-frame tools retain synchronous/polling
+behaviour and do not report MCP progress or cancellation. Use a direct blocking command call
+for an MCP-cancellable render.