mirror of
https://github.com/storytold/effectcraft.git
synced 2026-10-14 19:27:05 -04:00
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) <[email protected]>
This commit is contained in:
1 parent
176cddcdad
commit
8ed8a7e220
3 files changed
+123
-25
No files matched your search
@@ -102,28 +102,30 @@ impl McpServer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Serve until EOF on `input`. Input is read on its own thread so a long render can report
|
/// 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`).
|
/// progress, be cancelled and let other requests through while it runs (`long_job`). The
|
||||||
pub fn serve(&mut self, input: impl BufRead + Send, mut output: impl Write) -> std::io::Result<()> {
|
/// 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>>();
|
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.
|
// 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 read_error = std::sync::Arc::new(std::sync::Mutex::new(None::<std::io::Error>));
|
||||||
let served = std::thread::scope(|scope| {
|
let reader_error = read_error.clone();
|
||||||
let read_error = &read_error;
|
std::thread::Builder::new().name("mcp-input".into()).spawn(move || {
|
||||||
std::thread::Builder::new().name("mcp-input".into()).spawn_scoped(scope, move || {
|
for line in input.lines() {
|
||||||
for line in input.lines() {
|
let line = match line {
|
||||||
let line = match line {
|
Ok(line) => line,
|
||||||
Ok(line) => line,
|
Err(e) => {
|
||||||
Err(e) => {
|
*reader_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(e);
|
||||||
*read_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(e);
|
break;
|
||||||
break;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
if tx.send(Some(line)).is_err() {
|
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
};
|
||||||
|
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() {
|
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));
|
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 let Some(call) = long {
|
||||||
@@ -142,7 +144,7 @@ impl McpServer {
|
|||||||
Some(e) => Err(e),
|
Some(e) => Err(e),
|
||||||
None => Ok(()),
|
None => Ok(()),
|
||||||
}
|
}
|
||||||
});
|
})();
|
||||||
match (served, self.checkpoint(true)) {
|
match (served, self.checkpoint(true)) {
|
||||||
(r, Ok(())) => r,
|
(r, Ok(())) => r,
|
||||||
(Ok(()), Err(e)) => Err(e),
|
(Ok(()), Err(e)) => Err(e),
|
||||||
|
|||||||
@@ -211,7 +211,7 @@ fn stdio_loop() {
|
|||||||
.map(|v| v.to_string() + "\n")
|
.map(|v| v.to_string() + "\n")
|
||||||
.collect::<String>();
|
.collect::<String>();
|
||||||
let mut out = Vec::new();
|
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();
|
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.len(), 2);
|
||||||
assert_eq!(lines[1]["id"], 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"}}}}),
|
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>();
|
].iter().map(|v| v.to_string() + "\n").collect::<String>();
|
||||||
let mut output = Vec::new();
|
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();
|
let path = s.backend().session().unwrap().autosave.last_path.clone();
|
||||||
assert!(path.is_some(), "disconnect discarded the dirty project without an auto-save");
|
assert!(path.is_some(), "disconnect discarded the dirty project without an auto-save");
|
||||||
let path = path.unwrap();
|
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();
|
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]
|
#[test]
|
||||||
fn autosave_flushes_on_read_errors_and_broken_output() {
|
fn autosave_flushes_on_read_errors_and_broken_output() {
|
||||||
struct Broken;
|
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`
|
/// 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,
|
/// 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.
|
/// and `notifications/cancelled` stops it, deletes the partial file and sends no response.
|
||||||
|
|||||||
+20
-4
@@ -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
|
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
|
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
|
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.
|
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
|
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
|
`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.
|
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
|
`ping`, inspection and other requests are served while rendering. Closing stdin lets the
|
||||||
a second render while one is active. Closing stdin lets the pending render finish and reply.
|
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
|
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
|
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`.
|
contract. A failed queue item gives `isError: true`.
|
||||||
|
|
||||||
Bridge calls, `wait: false`, batch tools and single-frame tools retain synchronous/polling
|
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.
|
for an MCP-cancellable render.
|
||||||
Reference in new issue
Block a user