Merge pull request #458 from storybored379/fix/mcp-review-followups

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::<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 read_error = std::sync::Arc::new(std::sync::Mutex::new(None::<std::io::Error>));
+        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::<Value>(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),

crates/automation/src/tests.rs

@@ -229,7 +229,7 @@ fn stdio_loop() {
     .map(|v| v.to_string() + "\n")
     .collect::<String>();
     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<Value> = 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);
@@ -247,7 +247,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::<String>();
     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();
@@ -371,6 +371,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<usize> {
+            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;
@@ -917,6 +941,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.

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.