mirror of
https://github.com/storytold/effectcraft.git
synced 2026-10-24 10:09:32 -06:00
Merge pull request #365 from storybored379/feat/mcp-conventions
MCP: core tools, resources, render progress and cancellation
This commit is contained in:
@@ -19,6 +19,7 @@ mod autosave;
|
||||
pub mod backend;
|
||||
pub mod base64;
|
||||
pub mod bridge;
|
||||
mod long_job;
|
||||
pub mod server;
|
||||
pub mod tools;
|
||||
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
//! Long renders over MCP (docs/mcp.md "Progress and cancellation"). While [`McpServer::serve`]
|
||||
//! runs, a blocking `renderQueue.render` (`command_run`, headless) renders on the session's
|
||||
//! background render job instead of inline: the server keeps reading stdin, answers every other
|
||||
//! request in between (the job renders a snapshot of the project), reports progress as
|
||||
//! `notifications/progress` when the request carried `_meta.progressToken`, and stops the render
|
||||
//! on `notifications/cancelled` (deleting the partial output, sending no response).
|
||||
|
||||
use std::io::Write;
|
||||
use std::sync::mpsc::{Receiver, RecvTimeoutError};
|
||||
use std::time::Duration;
|
||||
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::server::{McpServer, call_result};
|
||||
use crate::tools::{self, Reply};
|
||||
|
||||
/// A `tools/call` result for `call`, with `resultType` for modern clients.
|
||||
fn result_for(call: &LongCall, r: Result<Reply, crate::Error>) -> Value {
|
||||
let mut result = call_result(r);
|
||||
if call.modern {
|
||||
result["resultType"] = json!("complete");
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// How often the job is polled (and at most how often progress is reported).
|
||||
const POLL: Duration = Duration::from_millis(100);
|
||||
|
||||
/// A `tools/call` that runs as a long job.
|
||||
pub struct LongCall {
|
||||
pub id: Value,
|
||||
pub params: Value,
|
||||
pub token: Option<Value>,
|
||||
/// The request declared MCP 2026-07-28+ (its result then carries `resultType`).
|
||||
pub modern: bool,
|
||||
}
|
||||
|
||||
/// Whether `msg` is a blocking `renderQueue.render` call (valid arguments, headless backend).
|
||||
pub fn long_call(server: &mut McpServer, msg: &Value) -> Option<LongCall> {
|
||||
let id = msg.get("id").filter(|i| i.is_string() || i.is_number())?;
|
||||
if msg.get("method").and_then(Value::as_str) != Some("tools/call") || server.backend().is_bridge() {
|
||||
return None;
|
||||
}
|
||||
let p = msg.get("params")?;
|
||||
let name = p.get("name")?.as_str()?;
|
||||
let args = p.get("arguments").cloned().unwrap_or(json!({}));
|
||||
if !matches!(name, "command_run" | "execute_command")
|
||||
|| args.get("id").or_else(|| args.get("command")).and_then(Value::as_str) != Some("renderQueue.render")
|
||||
{
|
||||
return None;
|
||||
}
|
||||
if tools::find(name).and_then(|t| t.unknown_arg(&args)).is_some() {
|
||||
return None; // the normal path reports it
|
||||
}
|
||||
let params = match args.get("params") {
|
||||
None | Some(Value::Null) => json!({}),
|
||||
Some(v) if v.is_object() => v.clone(),
|
||||
_ => return None,
|
||||
};
|
||||
if params.get("wait").and_then(Value::as_bool) == Some(false) {
|
||||
return None;
|
||||
}
|
||||
let token = p.get("_meta").and_then(|m| m.get("progressToken")).filter(|t| t.is_string() || t.is_number()).cloned();
|
||||
Some(LongCall { id: id.clone(), params, token, modern: crate::server::is_modern(p) })
|
||||
}
|
||||
|
||||
fn send(out: &mut impl Write, v: &Value) -> std::io::Result<()> {
|
||||
out.write_all(v.to_string().as_bytes())?;
|
||||
out.write_all(b"\n")?;
|
||||
out.flush()
|
||||
}
|
||||
|
||||
/// The `requestId` of a `notifications/cancelled` line.
|
||||
fn cancelled_id(line: &str) -> Option<Value> {
|
||||
let v: Value = serde_json::from_str(line).ok()?;
|
||||
(v.get("method")?.as_str()? == "notifications/cancelled").then(|| v.get("params")?.get("requestId").cloned())?
|
||||
}
|
||||
|
||||
impl McpServer {
|
||||
/// Run `call` to completion or cancellation, handling the other messages from `inbox`
|
||||
/// (`None` = stdin closed) meanwhile. Returns false when stdin closed.
|
||||
pub(crate) fn run_long(&mut self, call: LongCall, inbox: &Receiver<Option<String>>, out: &mut impl Write) -> std::io::Result<bool> {
|
||||
let mut params = call.params.clone();
|
||||
params["wait"] = json!(false);
|
||||
let start = self.guarded("command_run", &json!({"id": "renderQueue.render", "params": params}));
|
||||
let items = match &start {
|
||||
Ok(Reply::Json(v)) => v["items"].as_array().cloned().unwrap_or_default(),
|
||||
_ => {
|
||||
send(out, &json!({"jsonrpc": "2.0", "id": call.id, "result": result_for(&call, start)}))?;
|
||||
return Ok(true);
|
||||
}
|
||||
};
|
||||
let ids: Vec<u64> = items.iter().filter_map(|i| i["id"].as_u64()).collect();
|
||||
let shared = self.backend().session().and_then(|s| s.render_job.as_ref().map(|job| job.shared.clone()));
|
||||
let mut last = -1.0f64;
|
||||
let mut last_report: Option<std::time::Instant> = None;
|
||||
let mut open = true;
|
||||
let mut cancelled = false;
|
||||
loop {
|
||||
let state = shared.as_ref().map(|job| job.snapshot());
|
||||
let rendering = state.as_ref().is_some_and(|state| !state.finished);
|
||||
if let (Some(token), Some(st)) = (&call.token, &state)
|
||||
&& st.items_total > 0
|
||||
&& !cancelled
|
||||
{
|
||||
let frac = if st.total > 0 { st.done as f64 / st.total as f64 } else { 0.0 };
|
||||
// the current item's frames count until it is done
|
||||
let progress = (st.items_done as f64 + if st.items_done < st.items_total { frac.min(1.0) } else { 0.0 }).min(st.items_total as f64);
|
||||
if progress > last && (!rendering || last_report.is_none_or(|at| at.elapsed() >= POLL)) {
|
||||
last = progress;
|
||||
last_report = Some(std::time::Instant::now());
|
||||
let message = format!(
|
||||
"Rendering item {} of {}: frame {} of {}",
|
||||
st.items_done.saturating_add(1).min(st.items_total),
|
||||
st.items_total,
|
||||
st.done,
|
||||
st.total
|
||||
);
|
||||
send(
|
||||
out,
|
||||
&json!({"jsonrpc": "2.0", "method": "notifications/progress",
|
||||
"params": {"progressToken": token, "progress": progress, "total": st.items_total, "message": message}}),
|
||||
)?;
|
||||
}
|
||||
}
|
||||
if !rendering {
|
||||
break;
|
||||
}
|
||||
if !open {
|
||||
// stdin closed: finish the render (a piped script expects its reply), read nothing more
|
||||
std::thread::sleep(POLL);
|
||||
continue;
|
||||
}
|
||||
match inbox.recv_timeout(POLL) {
|
||||
Ok(Some(line)) => {
|
||||
if cancelled_id(&line).as_ref() == Some(&call.id) {
|
||||
cancelled = true;
|
||||
if let Some(job) = &shared {
|
||||
job.cancel.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if let Some(other) = serde_json::from_str::<Value>(&line).ok().and_then(|msg| long_call(self, &msg)) {
|
||||
let result = result_for(&other, Err(crate::Error::Other("a render is already in progress".into())));
|
||||
send(out, &json!({"jsonrpc":"2.0", "id":other.id, "result":result}))?;
|
||||
continue;
|
||||
}
|
||||
if let Some(reply) = self.handle_line(&line) {
|
||||
out.write_all(reply.as_bytes())?;
|
||||
out.write_all(b"\n")?;
|
||||
out.flush()?;
|
||||
}
|
||||
}
|
||||
Ok(None) | Err(RecvTimeoutError::Disconnected) => open = false,
|
||||
Err(RecvTimeoutError::Timeout) => {}
|
||||
}
|
||||
}
|
||||
let Some(s) = self.backend().session() else { return Ok(open) };
|
||||
s.poll_render();
|
||||
s.drain_events();
|
||||
let list = self.guarded("command_run", &json!({"id": "renderQueue.list"}));
|
||||
let items: Vec<Value> = match &list {
|
||||
Ok(Reply::Json(v)) => {
|
||||
v["items"].as_array().cloned().unwrap_or_default().into_iter().filter(|i| i["id"].as_u64().is_some_and(|id| ids.contains(&id))).collect()
|
||||
}
|
||||
_ => Vec::new(),
|
||||
};
|
||||
if cancelled {
|
||||
// Exporters remove only files they actually opened for this interrupted output.
|
||||
return Ok(open); // MCP: no response for a cancelled request
|
||||
}
|
||||
let failed = ids.len().saturating_sub(items.len()).saturating_add(items.iter().filter(|item| item["statusLabel"].as_str() != Some("Done")).count());
|
||||
let result = result_for(&call, Ok(Reply::Json(json!({"rendering": false, "items": items, "failed": failed}))));
|
||||
send(out, &json!({"jsonrpc": "2.0", "id": call.id, "result": result}))?;
|
||||
Ok(open)
|
||||
}
|
||||
}
|
||||
+114
-18
@@ -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;
|
||||
@@ -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 <port>`) 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 <port>`) 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),
|
||||
})
|
||||
}
|
||||
@@ -96,18 +101,48 @@ impl McpServer {
|
||||
Some(reply.to_string())
|
||||
}
|
||||
|
||||
/// Serve until EOF on `input`.
|
||||
pub fn serve(&mut self, input: impl BufRead, mut output: impl Write) -> std::io::Result<()> {
|
||||
let served = (|| {
|
||||
for line in input.lines() {
|
||||
if let Some(reply) = self.handle_line(&line?) {
|
||||
/// Serve until EOF on `input`. Input is read on its own thread so a long render can report
|
||||
/// progress, be cancelled and let other requests through while it runs (`long_job`).
|
||||
pub fn serve(&mut self, input: impl BufRead + Send, mut output: impl Write) -> std::io::Result<()> {
|
||||
let (tx, inbox) = std::sync::mpsc::channel::<Option<String>>();
|
||||
// A read error ends the input like EOF and is returned once the queued requests are answered.
|
||||
let read_error = std::sync::Mutex::new(None::<std::io::Error>);
|
||||
let served = std::thread::scope(|scope| {
|
||||
let read_error = &read_error;
|
||||
std::thread::Builder::new().name("mcp-input".into()).spawn_scoped(scope, move || {
|
||||
for line in input.lines() {
|
||||
let line = match line {
|
||||
Ok(line) => line,
|
||||
Err(e) => {
|
||||
*read_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(e);
|
||||
break;
|
||||
}
|
||||
};
|
||||
if tx.send(Some(line)).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
let _ = tx.send(None);
|
||||
})?;
|
||||
while let Ok(Some(line)) = inbox.recv() {
|
||||
let long = serde_json::from_str::<Value>(line.trim()).ok().and_then(|m| crate::long_job::long_call(self, &m));
|
||||
if let Some(call) = long {
|
||||
if !self.run_long(call, &inbox, &mut output)? {
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if let Some(reply) = self.handle_line(&line) {
|
||||
output.write_all(reply.as_bytes())?;
|
||||
output.write_all(b"\n")?;
|
||||
output.flush()?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
})();
|
||||
match read_error.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take() {
|
||||
Some(e) => Err(e),
|
||||
None => Ok(()),
|
||||
}
|
||||
});
|
||||
match (served, self.checkpoint(true)) {
|
||||
(r, Ok(())) => r,
|
||||
(Ok(()), Err(e)) => Err(e),
|
||||
@@ -117,14 +152,14 @@ impl McpServer {
|
||||
|
||||
/// Serve on stdin/stdout (the MCP stdio transport).
|
||||
pub fn serve_stdio(&mut self) -> std::io::Result<()> {
|
||||
self.serve(std::io::stdin().lock(), std::io::stdout().lock())
|
||||
self.serve(std::io::BufReader::new(std::io::stdin()), std::io::stdout().lock())
|
||||
}
|
||||
|
||||
fn request(&mut self, method: &str, params: &Value) -> Result<Value, (i64, String)> {
|
||||
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 +168,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 +181,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 +219,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 +246,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<Reply, Error>) -> 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 +260,47 @@ pub fn call_result(r: Result<Reply, Error>) -> 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<Reply, Error> {
|
||||
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::<String>().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}})
|
||||
}
|
||||
|
||||
@@ -734,3 +734,322 @@ 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<bool>);
|
||||
impl FootageSource for BrokenFootage {
|
||||
fn frame(&self, _: ItemId, _: &Footage, _: Tick) -> Option<Arc<Image>> {
|
||||
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);
|
||||
}
|
||||
|
||||
/// A render that takes ~2 ms per frame and writes its file first (so a cancel leaves a partial
|
||||
/// one to clean up).
|
||||
struct SlowExporter;
|
||||
|
||||
impl effectcraft_engine::Exporter for SlowExporter {
|
||||
fn formats(&self) -> Vec<effectcraft_engine::project::render_queue::OutputFormat> {
|
||||
effectcraft_engine::project::render_queue::OutputFormat::ALL.to_vec()
|
||||
}
|
||||
fn export(&self, job: &effectcraft_engine::ExportJob, progress: &mut dyn FnMut(u64, u64) -> bool) -> Result<effectcraft_engine::ExportResult, String> {
|
||||
std::fs::write(job.path, b"partial").map_err(|e| e.to_string())?;
|
||||
let n = 2000;
|
||||
for k in 0..=n {
|
||||
if !progress(k, n) {
|
||||
std::fs::remove_file(job.path).map_err(|e| e.to_string())?;
|
||||
return Err(effectcraft_engine::render_queue::CANCELLED.into());
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(2));
|
||||
}
|
||||
Ok(effectcraft_engine::ExportResult { path: job.path.to_string(), frames: n, ..Default::default() })
|
||||
}
|
||||
}
|
||||
|
||||
/// docs/mcp.md "Progress and cancellation" over the stdio loop: a blocking `renderQueue.render`
|
||||
/// reports `notifications/progress` for its token, other requests are answered while it renders,
|
||||
/// and `notifications/cancelled` stops it, deletes the partial file and sends no response.
|
||||
#[test]
|
||||
fn render_progress_and_cancel() {
|
||||
let dir = std::env::temp_dir().join(format!("effectcraft-mcp-progress-{}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let (in_r, mut in_w) = std::io::pipe().unwrap();
|
||||
let (out_r, out_w) = std::io::pipe().unwrap();
|
||||
let server = std::thread::spawn(move || {
|
||||
let session = Session { exporter: Some(std::sync::Arc::new(SlowExporter)), ..Default::default() };
|
||||
McpServer::new(Backend::headless(session)).serve(BufReader::new(in_r), out_w).unwrap();
|
||||
});
|
||||
let mut lines = BufReader::new(out_r).lines();
|
||||
let mut next = || -> Value { serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap() };
|
||||
let mut send = |v: Value| writeln!(in_w, "{v}").unwrap();
|
||||
let run = |id: u64, cmd: &str, params: Value, token: Option<Value>| {
|
||||
let mut p = json!({"name": "command_run", "arguments": {"id": cmd, "params": params}});
|
||||
if let Some(t) = token {
|
||||
p["_meta"] = json!({"progressToken": t, "io.modelcontextprotocol/protocolVersion":"2026-07-28"});
|
||||
}
|
||||
json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": p})
|
||||
};
|
||||
send(json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}));
|
||||
next();
|
||||
send(run(2, "comp.new", json!({"width": 64, "height": 64, "duration": 1}), None));
|
||||
assert_eq!(next()["result"]["isError"], false);
|
||||
let a = dir.join("a.mp4").to_string_lossy().to_string();
|
||||
send(run(3, "renderQueue.add", json!({"output": a}), None));
|
||||
assert_eq!(next()["result"]["isError"], false);
|
||||
// progress for the token, strictly increasing; doc_inspect answered while rendering
|
||||
send(run(4, "renderQueue.render", json!({}), Some(json!("tok"))));
|
||||
let (mut last, mut notes, mut asked, mut answered) = (-1.0, 0, false, false);
|
||||
let done = loop {
|
||||
let m = next();
|
||||
if m["method"] == "notifications/progress" {
|
||||
assert_eq!(m["params"]["progressToken"], "tok", "{m}");
|
||||
let p = m["params"]["progress"].as_f64().unwrap();
|
||||
assert!(p > last && p <= m["params"]["total"].as_f64().unwrap(), "{m}");
|
||||
(last, notes) = (p, notes + 1);
|
||||
if !asked && p > 0.0 {
|
||||
asked = true;
|
||||
send(json!({"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "doc_inspect", "arguments": {}}}));
|
||||
}
|
||||
} else if m["id"] == 5 {
|
||||
answered = true;
|
||||
assert_eq!(m["result"]["isError"], false, "{m}");
|
||||
} else {
|
||||
assert_eq!(m["id"], 4, "{m}");
|
||||
break m;
|
||||
}
|
||||
};
|
||||
assert_eq!(done["result"]["isError"], false, "{done}");
|
||||
assert_eq!(done["result"]["resultType"], "complete");
|
||||
assert_eq!(last, 1.0, "final progress reaches the total");
|
||||
let v: Value = serde_json::from_str(done["result"]["content"][0]["text"].as_str().unwrap()).unwrap();
|
||||
assert_eq!(v["items"][0]["statusLabel"], "Done", "{v}");
|
||||
assert!(notes >= 2 && answered, "{notes} notifications; doc_inspect answered during the render: {answered}");
|
||||
// cancel after the first notification: no response, the partial file is gone
|
||||
let b = dir.join("b.mp4");
|
||||
send(run(6, "renderQueue.add", json!({"output": b.to_string_lossy()}), None));
|
||||
assert_eq!(next()["result"]["isError"], false);
|
||||
send(run(7, "renderQueue.render", json!({}), Some(json!(7))));
|
||||
loop {
|
||||
let m = next();
|
||||
assert_ne!(m["id"], 7, "{m}");
|
||||
if m["method"] == "notifications/progress" && m["params"]["progress"].as_f64().unwrap() > 0.0 {
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert!(b.exists(), "the render started writing");
|
||||
send(json!({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 7, "reason": "test"}}));
|
||||
send(json!({"jsonrpc": "2.0", "id": 8, "method": "ping"}));
|
||||
let m = loop {
|
||||
let m = next();
|
||||
if m["method"] != "notifications/progress" {
|
||||
break m;
|
||||
}
|
||||
};
|
||||
assert_eq!(m["id"], 8, "no response for the cancelled request: {m}");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
|
||||
let v = loop {
|
||||
send(run(9, "renderQueue.list", json!({}), None));
|
||||
let v: Value = serde_json::from_str(next()["result"]["content"][0]["text"].as_str().unwrap()).unwrap();
|
||||
// the queue stops rendering a moment before the stopped item's status is updated
|
||||
if v["rendering"] == false && v["items"][1]["statusLabel"] != "Rendering" {
|
||||
break v;
|
||||
}
|
||||
assert!(std::time::Instant::now() < deadline, "cancel did not stop the job: {v}");
|
||||
std::thread::sleep(std::time::Duration::from_millis(20));
|
||||
};
|
||||
assert!(!b.exists(), "partial output deleted");
|
||||
assert!(std::path::Path::new(&a).exists(), "completed earlier queue item is retained");
|
||||
assert_eq!(v["items"][1]["statusLabel"], "User Stopped", "{v}");
|
||||
drop(send);
|
||||
drop(in_w);
|
||||
server.join().unwrap();
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn render_without_token_finishes_after_stdin_closes() {
|
||||
let dir = std::env::temp_dir().join(format!("ec-mcp-eof-{}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let path = dir.join("complete.mp4");
|
||||
let mut s = McpServer::new(Backend::headless(Session { exporter: Some(std::sync::Arc::new(SlowExporter)), ..Default::default() }));
|
||||
let mut input = String::new();
|
||||
for (id, command, params) in
|
||||
[(1, "comp.new", json!({"width":16,"height":16})), (2, "renderQueue.add", json!({"output":path})), (3, "renderQueue.render", json!({}))]
|
||||
{
|
||||
input.push_str(&format!(
|
||||
"{}\n",
|
||||
json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"command_run","arguments":{"id":command,"params":params}}})
|
||||
));
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
s.serve(std::io::Cursor::new(input), &mut out).unwrap();
|
||||
let replies: Vec<Value> = String::from_utf8(out).unwrap().lines().map(|line| serde_json::from_str(line).unwrap()).collect();
|
||||
assert_eq!(replies.len(), 3, "no token means no notifications: {replies:?}");
|
||||
assert_eq!(replies[2]["result"]["isError"], false);
|
||||
assert!(path.exists());
|
||||
std::fs::remove_dir_all(dir).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn panicked_render_worker_finishes_with_an_error() {
|
||||
struct BrokenExporter;
|
||||
impl effectcraft_engine::Exporter for BrokenExporter {
|
||||
fn formats(&self) -> Vec<effectcraft_engine::project::render_queue::OutputFormat> {
|
||||
effectcraft_engine::project::render_queue::OutputFormat::ALL.to_vec()
|
||||
}
|
||||
fn export(&self, _: &effectcraft_engine::ExportJob, _: &mut dyn FnMut(u64, u64) -> bool) -> Result<effectcraft_engine::ExportResult, String> {
|
||||
panic!("synthetic exporter failure");
|
||||
}
|
||||
}
|
||||
let mut session = Session { exporter: Some(std::sync::Arc::new(BrokenExporter)), ..Default::default() };
|
||||
session.execute("comp.new", json!({"width":16,"height":16})).unwrap();
|
||||
session.execute("renderQueue.add", json!({"output":"unused.mp4"})).unwrap();
|
||||
session.execute("renderQueue.render", json!({"wait":false})).unwrap();
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
|
||||
while session.is_rendering() && std::time::Instant::now() < deadline {
|
||||
std::thread::sleep(std::time::Duration::from_millis(10));
|
||||
}
|
||||
assert!(!session.is_rendering(), "a panicked worker must not leave an unfinishable job");
|
||||
session.poll_render();
|
||||
let queue = session.execute("renderQueue.list", json!({})).unwrap();
|
||||
assert!(queue.to_string().contains("synthetic exporter failure"), "{queue}");
|
||||
session.execute("renderQueue.add", json!({"output":"unused-again.mp4"})).unwrap();
|
||||
let mut server = McpServer::new(Backend::headless(session));
|
||||
let input =
|
||||
json!({"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"command_run","arguments":{"id":"renderQueue.render"}}}).to_string() + "\n";
|
||||
let mut out = Vec::new();
|
||||
server.serve(std::io::Cursor::new(input), &mut out).unwrap();
|
||||
let reply: Value = serde_json::from_slice(&out).unwrap();
|
||||
assert_eq!(reply["result"]["isError"], true, "{reply}");
|
||||
}
|
||||
|
||||
@@ -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<String> {
|
||||
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::<String>() + 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<Reply> {
|
||||
// ------------------------------------------------------------------ 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<Reply> {
|
||||
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<Reply> {
|
||||
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}))
|
||||
}
|
||||
|
||||
@@ -58,7 +58,8 @@ pub trait Exporter: Send + Sync {
|
||||
/// Formats this build can write.
|
||||
fn formats(&self) -> Vec<OutputFormat>;
|
||||
/// Render and write one item (blocking). `progress(done, total)` returns `false` to cancel;
|
||||
/// a cancelled export returns `Err("cancelled")`.
|
||||
/// a cancelled export removes the files it opened for that output and returns `Err("cancelled")`.
|
||||
/// Completed earlier queue items and unrelated files must be preserved.
|
||||
fn export(&self, job: &ExportJob, progress: &mut dyn FnMut(u64, u64) -> bool) -> Result<ExportResult, String>;
|
||||
/// [`Exporter::export`] as a future: renders whose GPU readbacks are deferred (a browser
|
||||
/// job worker's WebGPU device) await the device between a frame's passes. The default runs
|
||||
@@ -496,7 +497,22 @@ impl Session {
|
||||
let sh = shared.clone();
|
||||
let thread = std::thread::Builder::new()
|
||||
.name("render-queue".into())
|
||||
.spawn(move || effectcraft_render::passes::block_on(run_work(work, &sh, &mut |_| {})))
|
||||
.spawn(move || {
|
||||
let pending: Vec<u64> = work.items.iter().filter_map(|modules| modules.first().map(|(item, _)| item.id)).collect();
|
||||
if let Err(error) = crate::guard::guarded("render worker", || {
|
||||
effectcraft_render::passes::block_on(run_work(work, &sh, &mut |_| {}));
|
||||
Ok(())
|
||||
}) {
|
||||
let completed = lock(&sh.state).items_done;
|
||||
let mut updates = lock(&sh.updates);
|
||||
for id in pending.into_iter().skip(completed) {
|
||||
if !updates.iter().any(|u| matches!(u, ItemUpdate::Finished { id: done, .. } if *done == id)) {
|
||||
updates.push(ItemUpdate::Finished { id, status: RenderStatus::Failed(error.to_string()), seconds: 0.0, output: None });
|
||||
}
|
||||
}
|
||||
lock(&sh.state).finished = true;
|
||||
}
|
||||
})
|
||||
.map_err(|e| e.to_string())?;
|
||||
self.render_job = Some(RenderJob { shared, thread: Some(thread), remote: None });
|
||||
}
|
||||
|
||||
@@ -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)?;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -24,6 +24,7 @@ pub(crate) struct Cx<'a> {
|
||||
pub frame_times: Option<Mutex<Vec<(u64, f64)>>>,
|
||||
/// Files written to an overflow folder.
|
||||
pub overflowed: Mutex<Vec<String>>,
|
||||
written: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
impl<'a> std::ops::Deref for Cx<'a> {
|
||||
@@ -68,7 +69,7 @@ impl<'a> Cx<'a> {
|
||||
let mut project = job.project.clone();
|
||||
apply_overrides(&mut project, job);
|
||||
let frame_times = (job.options.log == RenderLog::PlusPerFrameInfo).then(|| Mutex::new(Vec::new()));
|
||||
Cx { job, project, frame_times, overflowed: Mutex::new(vec![]) }
|
||||
Cx { job, project, frame_times, overflowed: Mutex::new(vec![]), written: Mutex::new(vec![]) }
|
||||
}
|
||||
|
||||
pub fn comp(&self) -> Option<&Comp> {
|
||||
@@ -187,6 +188,30 @@ impl<'a> Cx<'a> {
|
||||
crate::fit(px, img.width, img.height, w, h)
|
||||
}
|
||||
|
||||
/// Open a file and remember ownership for cancellation cleanup (including overflow paths).
|
||||
pub fn create(&self, path: &str) -> crate::Result<out::Out<'_>> {
|
||||
let file = out::create(self.sink, path)?;
|
||||
if self.sink.is_none() {
|
||||
self.written.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push(path.to_owned());
|
||||
}
|
||||
Ok(file)
|
||||
}
|
||||
|
||||
pub fn remove_cancelled_outputs(&self) -> crate::Result<()> {
|
||||
let mut failure = None;
|
||||
for path in self.written.lock().unwrap_or_else(std::sync::PoisonError::into_inner).iter() {
|
||||
if let Err(e) = std::fs::remove_file(path)
|
||||
&& e.kind() != std::io::ErrorKind::NotFound
|
||||
{
|
||||
failure.get_or_insert_with(|| crate::ExportError::Io(format!("cannot remove cancelled output {path}: {e}")));
|
||||
}
|
||||
}
|
||||
match failure {
|
||||
Some(error) => Err(error),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Where a file goes: `path`, or (Use Storage Overflow) the first overflow folder with room
|
||||
/// for `bytes` when the quota hook says the output volume is full.
|
||||
pub fn place(&self, path: &str, bytes: u64) -> String {
|
||||
@@ -246,7 +271,7 @@ impl<'a> Cx<'a> {
|
||||
s.push_str(&format!(" Frame {} ({:.3} s): rendered in {:.1} ms\n", f0 + i as i64, t.seconds(), secs * 1000.0));
|
||||
}
|
||||
}
|
||||
let mut f = out::create(self.job.sink, &path).ok()?;
|
||||
let mut f = self.create(&path).ok()?;
|
||||
std::io::Write::write_all(&mut f, s.as_bytes()).ok()?;
|
||||
f.finish().ok()?;
|
||||
Some(path)
|
||||
|
||||
@@ -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![] })
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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 |
|
||||
@@ -97,6 +99,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
|
||||
|
||||
+56
@@ -0,0 +1,56 @@
|
||||
# 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"}}
|
||||
```
|
||||
|
||||
## 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.
|
||||
Reference in New Issue
Block a user