From 8ed8a7e2201f8ac311bb984472fd05abbc4baf44 Mon Sep 17 00:00:00 2001 From: storybored379 Date: Fri, 9 Oct 2026 10:26:57 -0700 Subject: [PATCH] MCP: return on write errors while stdin is open; document requests during a render Follow-ups from the #365 review: - serve() no longer hangs when a reply cannot be written: the stdin reader thread is detached instead of scoped, so a failed write returns at once rather than waiting for a read that only ends at EOF. serve() now takes a 'static reader (serve_stdio is unchanged). - A test pins down requests during a render: a second render is refused however it is started (blocking, wait: false, inside a batch), and edits apply to the project while the job keeps rendering its snapshot. - docs/mcp.md: isError is true for any JSON result with failed > 0; what requests mean mid-render; wait: false renders are polled with renderQueue.list. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/automation/src/server.rs | 40 ++++++++-------- crates/automation/src/tests.rs | 84 ++++++++++++++++++++++++++++++++- docs/mcp.md | 24 ++++++++-- 3 files changed, 123 insertions(+), 25 deletions(-) diff --git a/crates/automation/src/server.rs b/crates/automation/src/server.rs index ba8ba034..99b7ac61 100644 --- a/crates/automation/src/server.rs +++ b/crates/automation/src/server.rs @@ -102,28 +102,30 @@ impl McpServer { } /// 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<()> { + /// progress, be cancelled and let other requests through while it runs (`long_job`). The + /// reader thread is detached: when writing a reply fails, the server returns at once instead + /// of waiting for a read that may never complete. + pub fn serve(&mut self, input: impl BufRead + Send + 'static, mut output: impl Write) -> std::io::Result<()> { let (tx, inbox) = std::sync::mpsc::channel::>(); // 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::); - 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 read_error = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let reader_error = read_error.clone(); + std::thread::Builder::new().name("mcp-input".into()).spawn(move || { + for line in input.lines() { + let line = match line { + Ok(line) => line, + Err(e) => { + *reader_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); - })?; + } + let _ = tx.send(None); + })?; + let served = (|| { while let Ok(Some(line)) = inbox.recv() { let long = serde_json::from_str::(line.trim()).ok().and_then(|m| crate::long_job::long_call(self, &m)); if let Some(call) = long { @@ -142,7 +144,7 @@ impl McpServer { Some(e) => Err(e), None => Ok(()), } - }); + })(); match (served, self.checkpoint(true)) { (r, Ok(())) => r, (Ok(()), Err(e)) => Err(e), diff --git a/crates/automation/src/tests.rs b/crates/automation/src/tests.rs index 44a61cc1..b163f20b 100644 --- a/crates/automation/src/tests.rs +++ b/crates/automation/src/tests.rs @@ -211,7 +211,7 @@ fn stdio_loop() { .map(|v| v.to_string() + "\n") .collect::(); let mut out = Vec::new(); - s.serve(input.as_bytes(), &mut out).unwrap(); + s.serve(std::io::Cursor::new(input), &mut out).unwrap(); let lines: Vec = String::from_utf8(out).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect(); assert_eq!(lines.len(), 2); assert_eq!(lines[1]["id"], 2); @@ -229,7 +229,7 @@ fn headless_autosave_on_eof_preserves_unsaved_project() { json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "execute_command", "arguments": {"command": "layer.newText", "params": {"text": "Unsaved title"}}}}), ].iter().map(|v| v.to_string() + "\n").collect::(); let mut output = Vec::new(); - s.serve(input.as_bytes(), &mut output).unwrap(); + s.serve(std::io::Cursor::new(input), &mut output).unwrap(); let path = s.backend().session().unwrap().autosave.last_path.clone(); assert!(path.is_some(), "disconnect discarded the dirty project without an auto-save"); let path = path.unwrap(); @@ -351,6 +351,30 @@ fn autosave_marks_saved_or_undone_work_clean_without_losing_its_path() { std::fs::remove_dir_all(dir).unwrap(); } +/// A reply that cannot be written ends `serve` with the write error, even while stdin stays open +/// (the client is gone but never closed its end): the reader thread must not be waited for. +#[test] +fn broken_output_returns_while_stdin_stays_open() { + struct Broken; + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::ErrorKind::BrokenPipe.into()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let (in_r, mut in_w) = std::io::pipe().unwrap(); + writeln!(in_w, "{}", json!({"jsonrpc": "2.0", "id": 1, "method": "ping"})).unwrap(); + let (done_tx, done) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let _ = done_tx.send(server().serve(BufReader::new(in_r), Broken)); + }); + let result = done.recv_timeout(std::time::Duration::from_secs(10)).expect("serve must return while stdin is still open"); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::BrokenPipe); + drop(in_w); +} + #[test] fn autosave_flushes_on_read_errors_and_broken_output() { struct Broken; @@ -897,6 +921,62 @@ impl effectcraft_engine::Exporter for SlowExporter { } } +/// docs/mcp.md "Requests during a render": while a blocking render runs, every way of starting a +/// second render is refused (blocking, `wait: false`, inside a batch), and edits are applied to +/// the project without changing the running job, which renders the snapshot it started with. +#[test] +fn second_render_is_refused_and_edits_apply_during_a_render() { + let dir = std::env::temp_dir().join(format!("effectcraft-mcp-busy-{}", 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 call = + |id: u64, name: &str, arguments: Value| json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": {"name": name, "arguments": arguments}}); + send(json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}})); + next(); + send(call(2, "command_run", json!({"id": "comp.new", "params": {"width": 64, "height": 64, "duration": 1}}))); + assert_eq!(next()["result"]["isError"], false); + let a = dir.join("a.mp4").to_string_lossy().to_string(); + send(call(3, "command_run", json!({"id": "renderQueue.add", "params": {"output": a}}))); + assert_eq!(next()["result"]["isError"], false); + send(call(4, "command_run", json!({"id": "renderQueue.render"}))); + send(call(5, "command_run", json!({"id": "renderQueue.render", "params": {"wait": false}}))); + send(call(6, "command_run", json!({"id": "renderQueue.render"}))); + send(call(7, "command_batch", json!({"steps": [{"id": "renderQueue.render", "params": {"wait": false}}]}))); + send(call(8, "command_run", json!({"id": "comp.new", "params": {"name": "Added while rendering", "width": 32, "height": 32, "duration": 1}}))); + let mut replies = std::collections::HashMap::new(); + while !replies.contains_key(&4) { + let m = next(); + replies.insert(m["id"].as_u64().unwrap(), m); + } + for id in [5, 6, 7] { + let m = &replies[&id]; + assert_eq!(m["result"]["isError"], true, "a second render is refused: {m}"); + let text = m["result"]["content"][0]["text"].as_str().unwrap(); + assert!(text.contains("already in progress") || text.contains("is rendering"), "{m}"); + } + assert_eq!(replies[&8]["result"]["isError"], false, "edits are applied during a render"); + let done = &replies[&4]; + assert_eq!(done["result"]["isError"], false, "{done}"); + let v: Value = serde_json::from_str(done["result"]["content"][0]["text"].as_str().unwrap()).unwrap(); + assert_eq!(v["items"].as_array().unwrap().len(), 1, "{v}"); + assert_eq!(v["items"][0]["statusLabel"], "Done", "{v}"); + send(call(9, "doc_inspect", json!({}))); + let doc = next(); + assert!(doc.to_string().contains("Added while rendering"), "{doc}"); + drop(send); + drop(in_w); + server.join().unwrap(); + let _ = std::fs::remove_dir_all(&dir); +} + /// 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. diff --git a/docs/mcp.md b/docs/mcp.md index 9f10f403..0d7f83e9 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -22,7 +22,10 @@ open-world hints. Tools with an optional output path are conservatively annotate Unknown top-level tool argument keys return JSON-RPC `-32602` naming the key and accepted arguments. EffectCraft already rejects unknown engine command parameters; that behaviour is -preserved. Tool failures (including escaped panics) return `isError: true`. Malformed JSON +preserved. Tool failures (including escaped panics) return `isError: true`, and so does any JSON +result with a `failed` count above zero (a `command_batch` / `batch` with a failed step, a +render whose queue items did not all finish), even though the other steps or items succeeded: +read `results` / `items` for the details. Malformed JSON returns `-32700` with a null id and the session keeps serving. The MCP backend owns its session without a mutex; the engine's render-job mutexes already recover poisoned locks. @@ -41,8 +44,19 @@ Headless `command_run` / `execute_command` calls to `renderQueue.render` (unless 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. +`ping`, inspection and other requests are served while rendering. Closing stdin lets the +pending render finish and reply. + +### Requests during a render + +A render job owns a snapshot of the project taken when it started. While it runs: + +- Every way of starting a second render is refused with `isError: true`: a blocking + `renderQueue.render`, one with `wait: false`, or one inside a batch. +- Other commands, including edits, run normally and change the project. They do not affect the + running job, which keeps rendering its snapshot; to render an edit, render again after the + job ends. Queue items are matched by id when the job reports their status, so an item removed + mid-render is left out of the reply and counted in `failed`. 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 @@ -52,5 +66,7 @@ queue items and skipped existing frames remain. Custom exporters implement the s 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 +behaviour and do not report MCP progress or cancellation. `renderQueue.render {"wait": false}` +returns the ids being rendered at once; poll `renderQueue.list` for status and stop it with +`renderQueue.stop`. Use a direct blocking command call for an MCP-cancellable render.