From 164538fecb0c9db0feacf641cb79834030e07ea7 Mon Sep 17 00:00:00 2001 From: storybored379 Date: Thu, 8 Oct 2026 22:22:02 -0700 Subject: [PATCH 1/2] MCP: add core tools, annotations and resources Add strict tool keys, resource reads and panic recovery with regression coverage. Keep existing documented tools listed. Co-Authored-By: OpenAI Codex --- crates/automation/src/server.rs | 84 +++++++++++++++-- crates/automation/src/tests.rs | 140 +++++++++++++++++++++++++++++ crates/automation/src/tools.rs | 155 +++++++++++++++++++++++++++++++- docs/agents.md | 5 ++ docs/mcp.md | 36 ++++++++ 5 files changed, 408 insertions(+), 12 deletions(-) create mode 100644 docs/mcp.md diff --git a/crates/automation/src/server.rs b/crates/automation/src/server.rs index 899772a0..f0639354 100644 --- a/crates/automation/src/server.rs +++ b/crates/automation/src/server.rs @@ -13,13 +13,15 @@ use crate::{Backend, Error, base64}; /// Protocol revisions we speak, newest first. We answer with the client's if we know it. pub const PROTOCOL_VERSIONS: &[&str] = &["2025-06-18", "2025-03-26", "2024-11-05"]; -const INSTRUCTIONS: &str = "EffectCraft is an After Effects-class motion graphics compositor. Everything is an engine command: list_commands discovers ids and params, execute_command runs them (undoable). Typical flow: execute_command comp.new -> execute_command layer.newText / layer.newSolid (returns the layer id) -> set_property / add_keyframe -> execute_command effect.apply -> render_frame to look at the result. Inspect with get_project, get_comp and get_layer (every property node carries its `path`, e.g. `transform/position`, `effects/#1/blurriness`). Times are seconds. In bridge mode (app started with `--control `) screenshot and the ui_* tools show and operate the live window."; +const INSTRUCTIONS: &str = "EffectCraft is an After Effects-class motion graphics compositor. Everything is an engine command: command_list discovers ids and params, command_run {id, params} runs them (undoable; command_batch runs several in order). Typical flow: command_run comp.new -> command_run layer.newText / layer.newSolid (returns the layer id) -> set_property / add_keyframe -> command_run effect.apply -> render_preview to look at the result. Inspect with doc_inspect, get_project, get_comp and get_layer (every property node carries its `path`, e.g. `transform/position`, `effects/#1/blurriness`). Times are seconds. In bridge mode (app started with `--control `) screenshot and the ui_* tools show and operate the live window."; // JSON-RPC error codes. const PARSE_ERROR: i64 = -32700; 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; pub struct McpServer { backend: Backend, @@ -78,7 +80,10 @@ impl McpServer { }; let params = msg.get("params").cloned().unwrap_or(json!({})); Some(match self.request(method, ¶ms) { - Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}), + Ok(mut result) => { + modern_result_fields(method, ¶ms, &mut result); + json!({"jsonrpc": "2.0", "id": id, "result": result}) + } Err((code, m)) => error(id, code, &m), }) } @@ -124,7 +129,7 @@ impl McpServer { match method { "initialize" => { let asked = params.get("protocolVersion").and_then(Value::as_str).unwrap_or(""); - let version = PROTOCOL_VERSIONS.iter().find(|v| **v == asked).copied().unwrap_or(PROTOCOL_VERSIONS[0]); + let version = PROTOCOL_VERSIONS.iter().find(|v| **v == asked).copied().unwrap_or(PROTOCOL_VERSIONS.first().copied().unwrap_or("2025-06-18")); let mode = if self.backend.is_bridge() { "bridge" } else { "headless" }; let instructions = if self.autosave.is_some() { format!("{INSTRUCTIONS} Headless auto-save and recovery: {}", self.autosave_info()) @@ -133,7 +138,7 @@ impl McpServer { }; Ok(json!({ "protocolVersion": version, - "capabilities": {"tools": {"listChanged": false}}, + "capabilities": {"tools": {"listChanged": false}, "resources": {"listChanged": false, "subscribe": false}}, "serverInfo": {"name": "effectcraft", "title": format!("EffectCraft ({mode})"), "version": env!("CARGO_PKG_VERSION")}, "instructions": instructions, "_meta": {"effectcraftAutoSave": self.autosave_info()}, @@ -146,12 +151,15 @@ impl McpServer { } "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(json!({})); if tools::available(self.backend.is_bridge()).all(|t| t.name != name) { let hint = if tools::find(name).is_some() { format!(" ({})", crate::backend::NEED_BRIDGE) } else { String::new() }; return Err((INVALID_PARAMS, format!("unknown tool `{name}`{hint}"))); } - let args = params.get("arguments").cloned().unwrap_or(json!({})); - let mut result = call_result(tools::run(&mut self.backend, name, &args)); + if let Some(m) = tools::find(name).and_then(|t| t.unknown_arg(&args)) { + return Err((INVALID_PARAMS, m)); + } + let mut result = call_result(self.guarded(name, &args)); // Also checkpoint a partially successful tool that returned an error. The // project is durable before the client receives its reply (even if it kills // the process immediately afterwards). @@ -181,9 +189,24 @@ impl McpServer { } Ok(result) } - // Advertised as absent, but answer politely for clients that probe anyway. - "resources/list" => Ok(json!({"resources": []})), + "resources/list" => Ok(json!({"resources": [ + {"uri": DOCUMENT_URI, "name": "document", "title": "Project", "description": "The project overview and the active composition (same as doc_inspect).", "mimeType": "application/json"}, + {"uri": COMMANDS_URI, "name": "commands", "title": "Command catalog", "description": "Every engine command with id, label, menu, shortcut and params (same as command_list).", "mimeType": "application/json"}, + ]})), "resources/templates/list" => Ok(json!({"resourceTemplates": []})), + "resources/read" => { + let uri = params.get("uri").and_then(Value::as_str).ok_or((INVALID_PARAMS, "missing `uri`".to_string()))?; + let tool = match uri { + DOCUMENT_URI => "doc_inspect", + COMMANDS_URI => "command_list", + _ => return Err((RESOURCE_NOT_FOUND, format!("resource not found: {uri}"))), + }; + match self.guarded(tool, &json!({})) { + Ok(Reply::Json(v)) => Ok(json!({"contents": [{"uri": uri, "mimeType": "application/json", "text": v.to_string()}]})), + Ok(Reply::Image { .. }) => Err((INTERNAL_ERROR, "unexpected image".into())), + Err(e) => Err((INTERNAL_ERROR, e.to_string())), + } + } "prompts/list" => Ok(json!({"prompts": []})), m => Err((METHOD_NOT_FOUND, format!("method not found: {m}"))), } @@ -193,7 +216,9 @@ impl McpServer { /// A `tools/call` result: tool failures are reported in-band (`isError`) so the model can react. pub fn call_result(r: Result) -> Value { match r { - Ok(Reply::Json(v)) => json!({"content": [{"type": "text", "text": v.to_string()}], "isError": false}), + Ok(Reply::Json(v)) => { + json!({"content": [{"type": "text", "text": v.to_string()}], "isError": v.get("failed").and_then(Value::as_u64).is_some_and(|n| n > 0)}) + } Ok(Reply::Image { png, info }) => json!({ "content": [ {"type": "image", "data": base64::encode(&png), "mimeType": "image/png"}, @@ -205,6 +230,47 @@ pub fn call_result(r: Result) -> Value { } } +impl McpServer { + /// Run a tool with panics caught: a panic becomes an error result and the session keeps + /// serving (docs/mcp.md). + pub(crate) fn guarded(&mut self, name: &str, args: &Value) -> Result { + let backend = &mut self.backend; + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| tools::run(backend, name, args))).unwrap_or_else(|p| { + let msg = p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::().cloned()).unwrap_or_else(|| "panic".into()); + Err(Error::Other(format!("internal error: {msg}"))) + }) + } +} + +/// The craft catalog resources (docs/mcp.md). +const DOCUMENT_URI: &str = "effectcraft://document"; +const COMMANDS_URI: &str = "effectcraft://commands"; + +/// Whether a request declares MCP 2026-07-28 or later in its `_meta` (a "modern" client, which +/// negotiates per request instead of through `initialize`). +pub(crate) fn is_modern(params: &Value) -> bool { + params.pointer("/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Value::as_str).is_some_and(|v| v >= "2026-07-28") +} + +/// Modern clients reject list/read results without `resultType`, `ttlMs` and `cacheScope`; those +/// fields are added for them. Sessions that negotiated an older revision through `initialize` keep +/// the legacy shape. The tool and resource lists never change while the server runs (cacheable +/// for ten minutes); reads follow the project, so they are never cached. +pub(crate) fn modern_result_fields(method: &str, params: &Value, result: &mut Value) { + let ttl_ms = match method { + "tools/list" | "resources/list" | "resources/templates/list" | "prompts/list" => Some(600_000), + "resources/read" => Some(0), + "tools/call" => None, + _ => return, + }; + let Some(obj) = result.as_object_mut().filter(|_| is_modern(params)) else { return }; + obj.insert("resultType".into(), json!("complete")); + if let Some(ttl_ms) = ttl_ms { + obj.insert("ttlMs".into(), json!(ttl_ms)); + obj.insert("cacheScope".into(), json!("private")); + } +} + fn error(id: Value, code: i64, message: &str) -> Value { json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}) } diff --git a/crates/automation/src/tests.rs b/crates/automation/src/tests.rs index 7bea5004..263765ba 100644 --- a/crates/automation/src/tests.rs +++ b/crates/automation/src/tests.rs @@ -734,3 +734,143 @@ fn autosave_non_utf8_path_returns_an_error_instead_of_panicking() { let result = server().with_autosave(&root); assert!(matches!(result, Err(crate::Error::Other(message)) if message.contains("UTF-8"))); } + +#[test] +fn core_tools_annotations_and_strict_keys() { + let mut s = server(); + let tools = rpc(&mut s, 1, "tools/list", json!({})); + let tools = tools["tools"].as_array().unwrap(); + for name in ["command_list", "command_run", "command_batch", "doc_inspect", "render_preview"] { + assert!(tools.iter().any(|t| t["name"] == name), "missing {name}"); + } + for t in tools { + assert!(t["title"].as_str().is_some_and(|s| !s.is_empty())); + for hint in ["readOnlyHint", "destructiveHint", "idempotentHint", "openWorldHint"] { + assert!(t["annotations"][hint].is_boolean(), "{t}"); + } + let msg = json!({"jsonrpc":"2.0", "id":2, "method":"tools/call", "params":{"name":t["name"], "arguments":{"typo_key":1}}}); + let reply = s.handle(&msg).unwrap(); + assert_eq!(reply["error"]["code"], -32602, "{reply}"); + assert!(reply["error"]["message"].as_str().unwrap().contains("typo_key")); + } + let r = call_json(&mut s, "command_run", json!({"id":"comp.new", "params":{"width":32,"height":32}})); + assert!(r.is_object()); + let (content, err) = call(&mut s, "command_batch", json!({"steps":[{"id":"no.such.command"},{"id":"layer.newNull"}]})); + assert!(err); + let result: Value = serde_json::from_str(content[0]["text"].as_str().unwrap()).unwrap(); + assert_eq!(result["completed"], 0); + assert_eq!(result["failed"], 1); + assert_eq!(result["results"].as_array().unwrap().len(), 1); + let (content, err) = call(&mut s, "command_batch", json!({"stop_on_error":false,"steps":[{"id":"no.such.command"},{"id":"layer.newNull"}]})); + assert!(err); + let result: Value = serde_json::from_str(content[0]["text"].as_str().unwrap()).unwrap(); + assert_eq!(result["completed"], 1); + assert_eq!(result["failed"], 1); + assert!(call_json(&mut s, "doc_inspect", json!({}))["activeComp"].is_object()); + assert!(!call(&mut s, "render_preview", json!({"max_side":16})).1); +} + +#[test] +fn resources_match_tools_and_recover_after_parse_error() { + let mut s = server(); + let bad: Value = serde_json::from_str(&s.handle_line("{broken").unwrap()).unwrap(); + assert_eq!(bad["error"]["code"], -32700); + assert!(bad["id"].is_null()); + assert_eq!(rpc(&mut s, 1, "ping", json!({})), json!({})); + let list = rpc(&mut s, 2, "resources/list", json!({})); + for (uri, tool) in [("effectcraft://document", "doc_inspect"), ("effectcraft://commands", "command_list")] { + assert!(list["resources"].as_array().unwrap().iter().any(|r| r["uri"] == uri)); + let r = rpc(&mut s, 3, "resources/read", json!({"uri":uri})); + assert_eq!(r["contents"][0]["mimeType"], "application/json"); + let value: Value = serde_json::from_str(r["contents"][0]["text"].as_str().unwrap()).unwrap(); + assert_eq!(value, call_json(&mut s, tool, json!({}))); + } +} + +/// Rendering is a direct tool call, outside Session::execute's existing panic guard. +#[test] +fn render_panic_is_an_error_and_session_remains_usable() { + use effectcraft_engine::{ + color::Label, + project::{Footage, FootageKind, ItemId, ItemKind}, + raster::Image, + render::FootageSource, + time::Tick, + }; + use std::sync::{Arc, Mutex}; + struct BrokenFootage(Mutex); + impl FootageSource for BrokenFootage { + fn frame(&self, _: ItemId, _: &Footage, _: Tick) -> Option> { + let mut first = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if *first { + *first = false; + panic!("synthetic decoder failure"); + } + Some(Arc::new(Image::new(16, 16))) + } + } + let mut session = Session { footage: Arc::new(BrokenFootage(Mutex::new(true))), ..Default::default() }; + session.execute("comp.new", json!({"width":16,"height":16})).unwrap(); + let item = Arc::make_mut(&mut session.project).add_item( + "synthetic", + Label::Aqua, + None, + ItemKind::Footage(Footage { path: "synthetic.png".into(), kind: FootageKind::Still, width: 16, height: 16, has_video: true, ..Default::default() }), + ); + session.execute("layer.addItem", json!({"item":item.0})).unwrap(); + let mut s = McpServer::new(Backend::headless(session)); + let (content, err) = call(&mut s, "render_frame", json!({})); + assert!(err); + assert!(content[0]["text"].as_str().unwrap().contains("synthetic decoder failure")); + assert_eq!(rpc(&mut s, 2, "ping", json!({})), json!({})); + assert!(!call(&mut s, "render_frame", json!({})).1); +} + +/// MCP 2026-07-28 ("modern") clients declare their revision per request in `_meta` and reject +/// list/read results without `resultType`, `ttlMs` and `cacheScope`; legacy sessions (revision +/// negotiated through `initialize`) keep the old result shape. +#[test] +fn modern_clients_get_result_type() { + let mut s = server(); + let meta = json!({ + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": {"name": "t", "version": "1"}, + "io.modelcontextprotocol/clientCapabilities": {}, + }); + for (id, method, params) in [ + (1, "tools/list", json!({"_meta": meta})), + (2, "resources/list", json!({"_meta": meta})), + (3, "resources/templates/list", json!({"_meta": meta})), + (4, "resources/read", json!({"_meta": meta, "uri": "effectcraft://document"})), + (5, "tools/call", json!({"_meta": meta, "name": "command_list", "arguments": {}})), + ] { + let r = rpc(&mut s, id, method, params); + assert_eq!(r["resultType"], "complete", "{method}: {r}"); + if method != "tools/call" { + assert!(r["ttlMs"].is_u64(), "{method}: {r}"); + assert!(matches!(r["cacheScope"].as_str(), Some("public" | "private")), "{method}: {r}"); + } + } + let mut s = server(); + rpc(&mut s, 1, "initialize", json!({"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}})); + for (id, method, params) in + [(2, "tools/list", json!({})), (3, "resources/list", json!({})), (4, "resources/read", json!({"uri": "effectcraft://document"}))] + { + let r = rpc(&mut s, id, method, params); + let res = r.as_object().unwrap(); + assert!(!res.contains_key("resultType") && !res.contains_key("ttlMs") && !res.contains_key("cacheScope"), "{method}: {r}"); + } +} + +#[test] +fn render_job_state_recovers_a_poisoned_lock() { + let shared = std::sync::Arc::new(effectcraft_engine::render_queue::JobShared::default()); + let worker = shared.clone(); + let _ = std::thread::spawn(move || { + let mut state = worker.state.lock().unwrap(); + state.done = 3; + panic!("synthetic worker panic while holding the progress lock"); + }) + .join(); + assert_eq!(shared.snapshot().done, 3); +} diff --git a/crates/automation/src/tools.rs b/crates/automation/src/tools.rs index 87315767..89b8b7b8 100644 --- a/crates/automation/src/tools.rs +++ b/crates/automation/src/tools.rs @@ -31,10 +31,62 @@ impl ToolDef { pub fn input_schema(&self) -> Value { (self.schema)() } - /// The MCP `Tool` object. + /// The MCP `Tool` object, with a title and annotations (docs/mcp.md). pub fn descriptor(&self) -> Value { - json!({"name": self.name, "description": self.description, "inputSchema": self.input_schema()}) + let title = title_of(self.name); + let read_only = READ_ONLY.contains(&self.name); + let files = WRITES_FILES.contains(&self.name); + json!({ + "name": self.name, + "title": title, + "description": self.description, + "inputSchema": self.input_schema(), + "annotations": { + "title": title, + "readOnlyHint": read_only, + "destructiveHint": !read_only && !files, + "idempotentHint": read_only || files, + "openWorldHint": false, + }, + }) } + + /// The first argument this tool doesn't take, as a message naming the accepted ones. + pub fn unknown_arg(&self, args: &Value) -> Option { + let Some(m) = args.as_object() else { return Some("arguments must be an object".into()) }; + let schema = self.input_schema(); + let props = schema.get("properties")?.as_object()?; + let bad = m.keys().find(|k| !props.contains_key(*k))?; + let accepted: Vec<&str> = props.keys().map(String::as_str).collect(); + Some(format!("unknown argument \"{bad}\" for {}; expected: {}", self.name, accepted.join(", "))) + } +} + +/// Tools that only read. +const READ_ONLY: &[&str] = &[ + "command_list", + "list_commands", + "describe_command", + "doc_inspect", + "render_preview", + "get_project", + "get_comp", + "get_layer", + "get_property", + "list_effects", + "list_fonts", + "get_state", + "ui_inspect", + "ui_elements", +]; +/// Tools that write files without changing the project. +const WRITES_FILES: &[&str] = &["save_project", "render_frame", "screenshot"]; + +/// `get_project` → "Get project". +fn title_of(name: &str) -> String { + let s = name.replace('_', " "); + let mut c = s.chars(); + c.next().map(|f| f.to_uppercase().collect::() + c.as_str()).unwrap_or_default() } /// All tools (headless + bridge-only). @@ -64,7 +116,7 @@ pub fn run(b: &mut Backend, name: &str, args: &Value) -> Result { // ------------------------------------------------------------------ schema fragments fn schema(props: Value, required: &[&str]) -> Value { - json!({"type": "object", "properties": props, "required": required}) + json!({"type": "object", "properties": props, "required": required, "additionalProperties": false}) } fn comp_s() -> Value { json!({"type": ["integer", "string"], "description": "Composition item id or name; omit for the active comp."}) @@ -825,4 +877,101 @@ static TOOLS: &[ToolDef] = &[ }, run: batch_tool, }, + ToolDef { + name: "command_list", + description: "List engine commands: id, label, menu path, shortcut, `params` doc (e.g. `{name?, width?, height?, frameRate?, duration? (s)}`) and whether it can run now (`why` if not). Every menu item, shortcut and panel gesture maps to one of these ids; run them with execute_command. Narrow with `filter` (e.g. `layer.new`, `keys`, `effect`).", + bridge_only: false, + schema: || { + schema( + json!({ + "filter": {"type": "string", "description": "Only commands whose id or label contains this text (case-insensitive)."}, + "enabled_only": {"type": "boolean", "description": "Only commands that can run right now."}, + "schemas": {"type": "boolean", "description": "Also return each command's parameter JSON Schema (`schema`)."} + }), + &[], + ) + }, + run: list_commands, + }, + ToolDef { + name: "command_run", + description: "Run any engine command by id with JSON params; edits are undoable. Examples: comp.new {\"name\":\"Main\",\"width\":1920,\"height\":1080,\"frameRate\":30,\"duration\":5}; layer.newText {\"text\":\"Hello\",\"size\":120,\"fill\":\"#ffffff\"} (returns {\"layer\":id}); layer.newSolid {\"color\":\"#3366ff\"}; layer.newShape; effect.apply {\"layer\":3,\"effect\":\"Gaussian Blur\"}; time.set {\"time\":2.5}; layer.select {\"layers\":[3]}. Layers are referenced by id, \"#n\" or name; times are seconds.", + bridge_only: false, + schema: || { + schema( + json!({ + "id": {"type": "string", "description": "Command id, e.g. `comp.new`, `layer.newText`, `effect.apply` (see list_commands)."}, + "params": {"type": "object", "description": "Command parameters as documented by list_commands.", "additionalProperties": true} + }), + &["id"], + ) + }, + run: execute_command, + }, + ToolDef { + name: "render_preview", + description: "Render a composition frame (default: the active comp at the current time) and return it as a PNG image, longest side `max_side` (default 960, 0 = full size). Use it to check your work.", + bridge_only: false, + schema: || { + schema( + json!({ + "comp": comp_s(), + "time": {"type": "number", "description": "Comp time in seconds (default: current time)."}, + "max_side": {"type": "integer", "description": "Longest side in pixels (default 960; 0 = full comp size)."}, + "transparent": {"type": "boolean", "description": "Keep the frame's alpha (RGBA PNG, as an RGB + Alpha render writes it) instead of compositing over the comp's background colour."} + }), + &[], + ) + }, + run: render_frame, + }, + ToolDef { + name: "command_batch", + description: "Run commands in order, returning each result or error. Stop at the first error by default; stop_on_error: false continues. Each edit has its own undo step; use batch for an atomic undo group.", + bridge_only: false, + schema: || { + schema( + json!({"steps": {"type": "array", "items": {"type": "object", "properties": {"id": {"type": "string"}, "params": {"type": "object"}}, "required": ["id"], "additionalProperties": false}}, "stop_on_error": {"type": "boolean"}}), + &["steps"], + ) + }, + run: command_batch, + }, + ToolDef { + name: "doc_inspect", + description: "Inspect the project overview and active composition (null when no composition is active).", + bridge_only: false, + schema: || schema(json!({}), &[]), + run: doc_inspect, + }, ]; + +fn doc_inspect(b: &mut Backend, _: &Value) -> Result { + let project = b.exec("project.summary", json!({}))?; + let comp = b.exec("comp.info", json!({})).ok(); + json_reply(json!({"project": project, "activeComp": comp})) +} + +fn command_batch(b: &mut Backend, a: &Value) -> Result { + let steps = need(a, "steps")?.as_array().ok_or_else(|| Error::BadArgs("steps must be an array".into()))?; + let stop = get(a, "stop_on_error").and_then(Value::as_bool).unwrap_or(true); + let mut results = Vec::new(); + let (mut completed, mut failed) = (0, 0); + for step in steps { + match execute_command(b, step) { + Ok(Reply::Json(result)) => { + completed += 1; + results.push(json!({"ok": true, "result": result})); + } + Ok(Reply::Image { .. }) => return Err(Error::Other("unexpected command image".into())), + Err(e) => { + failed += 1; + results.push(json!({"ok": false, "error": e.to_string()})); + if stop { + break; + } + } + } + } + json_reply(json!({"completed": completed, "failed": failed, "results": results})) +} diff --git a/docs/agents.md b/docs/agents.md index 373f1b7d..29a9f589 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -97,6 +97,11 @@ may be absent from the last checkpoint. | `script_ui {action?, window?, widget?, value?, result?}` | Drive ScriptUI windows that scripts opened: `list` (default) the open windows, `get` a window's control tree, `click` a button / checkbox / radio button, `set` text, a slider or a list item, `close` a window (`result`: what a dialog's `show()` returns, default 2 = Cancel). | | `batch {steps: [{command, params}], label?, atomic?}` | Several commands in one call and one undo step (`engine.batch`); `"$N.key"` in a param is step N's result (`"$1.layer"`). A failing step rolls the batch back and names the step. | | *(bridge)* `screenshot {panel?, id?, max_side?, path?}`, `ui_inspect`, `ui_elements {prefix?}`, `ui_click`, `ui_drag`, `ui_key`, `ui_type`, `ui_set`, `control {method, params}` | Look at and operate the live window. | +| `command_list` | List commands (same arguments as `list_commands`). | +| `command_run {id, params?}` | Run a command. | +| `command_batch {steps, stop_on_error?}` | Run commands in order with per-step results; each edit has its own undo step. | +| `doc_inspect` | Project overview and active composition. | +| `render_preview` | Render a PNG without writing a file (`comp`, `time`, `max_side`, `transparent`). | Layers are referenced by id (from `get_comp`), `"#n"` (1-based index from the top) or name. Comps are referenced by id or name, and default to the active comp. Property paths come from `get_layer`, for diff --git a/docs/mcp.md b/docs/mcp.md new file mode 100644 index 00000000..855e04a3 --- /dev/null +++ b/docs/mcp.md @@ -0,0 +1,36 @@ +# MCP conventions + +Run `effectcraft-cli mcp` for a headless session, or `effectcraft-cli mcp --bridge 9877` +to use the desktop app's control channel. Existing ports are unchanged. + +The server provides the same core conventions as FilmCraft #28: + +| Tool | Arguments | Result | +| --- | --- | --- | +| `command_list` | `filter?`, `enabled_only?`, `schemas?` | Command catalog | +| `command_run` | `id`, `params?` | Engine command result | +| `command_batch` | `steps: [{id, params?}]`, `stop_on_error?` | `completed`, `failed`, per-step `results` | +| `doc_inspect` | none | Project overview and active composition | +| `render_preview` | `comp?`, `time?`, `max_side?`, `transparent?` | PNG and frame information | + +`command_batch` stops at the first error by default; false continues. Each edit has its +own undo step. The existing `batch` tool retains its atomic undo grouping and result references. +Existing tools, including `list_commands`, `execute_command`, and `batch`, remain listed +because existing workflows use them; there are no hidden aliases. See [agents.md](agents.md) +for the full catalog. Each tool has a title and read-only, destructive, idempotent and +open-world hints. Tools with an optional output path are conservatively annotated as file writers. + +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 +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. + +Resources `effectcraft://document` and `effectcraft://commands` return JSON matching +`doc_inspect` and `command_list`. Clients declaring MCP 2026-07-28 in per-request `_meta` +receive `resultType: complete` and list/read cache hints; document reads are never cached. + +```json +{"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"}} +``` From 7b632bdbc19fafc3962a1b0965aaeeb492587006 Mon Sep 17 00:00:00 2001 From: storybored379 Date: Thu, 8 Oct 2026 22:22:16 -0700 Subject: [PATCH 2/2] 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 Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/automation/src/lib.rs | 1 + crates/automation/src/long_job.rs | 177 ++++++++++++++++++++++++ crates/automation/src/server.rs | 48 +++++-- crates/automation/src/tests.rs | 179 +++++++++++++++++++++++++ crates/engine/src/render_queue.rs | 20 ++- crates/export/src/encode.rs | 2 +- crates/export/src/lib.rs | 8 +- crates/export/src/pipeline.rs | 29 +++- crates/export/src/webm.rs | 4 +- crates/export/src/webm_av1.rs | 2 +- crates/export/tests/render_settings.rs | 58 ++++++++ docs/agents.md | 2 + docs/mcp.md | 20 +++ 13 files changed, 531 insertions(+), 19 deletions(-) create mode 100644 crates/automation/src/long_job.rs diff --git a/crates/automation/src/lib.rs b/crates/automation/src/lib.rs index 9f38516e..6a673309 100644 --- a/crates/automation/src/lib.rs +++ b/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; diff --git a/crates/automation/src/long_job.rs b/crates/automation/src/long_job.rs new file mode 100644 index 00000000..4cf85824 --- /dev/null +++ b/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) -> 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, + /// 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 { + 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 { + 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>, out: &mut impl Write) -> std::io::Result { + 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 = 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 = 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::(&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 = 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) + } +} diff --git a/crates/automation/src/server.rs b/crates/automation/src/server.rs index f0639354..ba8ba034 100644 --- a/crates/automation/src/server.rs +++ b/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::>(); + // 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 _ = tx.send(None); + })?; + 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 { + 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 { diff --git a/crates/automation/src/tests.rs b/crates/automation/src/tests.rs index 263765ba..5853f6b9 100644 --- a/crates/automation/src/tests.rs +++ b/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::ALL.to_vec() + } + fn export(&self, job: &effectcraft_engine::ExportJob, progress: &mut dyn FnMut(u64, u64) -> bool) -> Result { + 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| { + 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 = 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::ALL.to_vec() + } + fn export(&self, _: &effectcraft_engine::ExportJob, _: &mut dyn FnMut(u64, u64) -> bool) -> Result { + 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}"); +} diff --git a/crates/engine/src/render_queue.rs b/crates/engine/src/render_queue.rs index 0ea4aac9..9829f861 100644 --- a/crates/engine/src/render_queue.rs +++ b/crates/engine/src/render_queue.rs @@ -58,7 +58,8 @@ pub trait Exporter: Send + Sync { /// Formats this build can write. fn formats(&self) -> Vec; /// 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; /// [`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 = 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 }); } diff --git a/crates/export/src/encode.rs b/crates/export/src/encode.rs index b42ec6a2..67ff2736 100644 --- a/crates/export/src/encode.rs +++ b/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)?; diff --git a/crates/export/src/lib.rs b/crates/export/src/lib.rs index 81e0f5c2..3a82d6ad 100644 --- a/crates/export/src/lib.rs +++ b/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); diff --git a/crates/export/src/pipeline.rs b/crates/export/src/pipeline.rs index 3b1e7161..3680ffcd 100644 --- a/crates/export/src/pipeline.rs +++ b/crates/export/src/pipeline.rs @@ -24,6 +24,7 @@ pub(crate) struct Cx<'a> { pub frame_times: Option>>, /// Files written to an overflow folder. pub overflowed: Mutex>, + written: Mutex>, } 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> { + 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) diff --git a/crates/export/src/webm.rs b/crates/export/src/webm.rs index 98993b9e..3cec2d62 100644 --- a/crates/export/src/webm.rs +++ b/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![] }) diff --git a/crates/export/src/webm_av1.rs b/crates/export/src/webm_av1.rs index f9b053b8..ca06a3e5 100644 --- a/crates/export/src/webm_av1.rs +++ b/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); diff --git a/crates/export/tests/render_settings.rs b/crates/export/tests/render_settings.rs index dfa62a7f..b956caa1 100644 --- a/crates/export/tests/render_settings.rs +++ b/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(); +} diff --git a/docs/agents.md b/docs/agents.md index 29a9f589..3fc19630 100644 --- a/docs/agents.md +++ b/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 | diff --git a/docs/mcp.md b/docs/mcp.md index 855e04a3..9f10f403 100644 --- a/docs/mcp.md +++ b/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.