MCP: progress notifications and cancellation for blocking exports

command_run file.exportMedia with wait: true (headless) runs as the
background export job and uses the MCP progress and cancellation utilities:

- with _meta.progressToken, notifications/progress reports frames done /
  total with the job status, at most every 100 ms (none without a token);
- notifications/cancelled stops the job at the next batch (jobs.cancel),
  deletes the partial output and sends no response;
- the session lock is only held for short polls, so ping, doc_inspect,
  jobs.list etc. are answered while it encodes.

Without wait, and in bridge mode, nothing changes. Test
export_progress_and_cancel.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
storybored379 2026-10-04 11:08:45 -07:00
parent 8ee7658303
commit cad16825e3
5 changed files with 236 additions and 2 deletions

View File

@ -9,6 +9,7 @@
//! and type in the live UI — every menu, panel, timeline gesture and keystroke.
pub mod bridge;
pub mod long_job;
pub mod server;
pub use bridge::BridgeClient;

View File

@ -0,0 +1,126 @@
//! Long exports over MCP (docs/agents.md § Long exports): `file.exportMedia` with `wait: true`
//! runs as a background engine job; the server reports its progress as MCP
//! `notifications/progress` (when the request carried a `progressToken`) and stops it on
//! `notifications/cancelled`, deleting the partial output. The session lock is only held for
//! short polls, so other requests are answered while the export runs.
use std::path::Path;
use std::time::{Duration, SystemTime};
use rmcp::RoleServer;
use rmcp::model::{CallToolResult, ContentBlock as Content, ProgressNotificationParam, ProgressToken};
use rmcp::service::{Peer, RequestContext};
use serde_json::{Value, json};
use crate::server::FilmcraftMcp;
/// Engine commands that block until a render is written when called with `wait: true`.
pub const LONG_COMMANDS: &[&str] = &["file.exportMedia"];
/// How often the job is polled (and at most how often progress is reported).
const POLL: Duration = Duration::from_millis(100);
/// Whether `command_run {id, params}` is a long call this module runs.
pub fn is_long(id: &str, params: &Value) -> bool {
LONG_COMMANDS.contains(&id) && params.get("wait").and_then(Value::as_bool) == Some(true)
}
/// Sends `notifications/progress` for one request; progress only ever increases.
struct Reporter {
peer: Peer<RoleServer>,
token: Option<ProgressToken>,
last: f64,
}
impl Reporter {
async fn report(&mut self, done: f64, total: f64, message: &str) {
let Some(token) = self.token.clone() else { return };
if total <= 0.0 || done <= self.last {
return;
}
self.last = done;
let mut p = ProgressNotificationParam::new(token, done).with_total(total);
if !message.is_empty() {
p = p.with_message(message);
}
let _ = self.peer.notify_progress(p).await;
}
}
impl FilmcraftMcp {
/// Run `id` (a [`LONG_COMMANDS`] entry, `wait: true`) as a job, reporting progress and
/// honouring cancellation.
pub(crate) async fn run_long(&self, id: &str, mut params: Value, context: &RequestContext<RoleServer>) -> CallToolResult {
let started = SystemTime::now();
if let Some(p) = params.as_object_mut() {
p.insert("wait".into(), json!(false));
}
let start = match self.run(id, params).await {
Ok(v) => v,
Err(e) => return CallToolResult::error(vec![Content::text(e.to_string())]),
};
let Some(job) = start.get("job").and_then(Value::as_u64) else {
return CallToolResult::error(vec![Content::text(format!("{id} started no job: {start}"))]);
};
let path = start.get("path").and_then(Value::as_str).unwrap_or_default().to_string();
let mut rep = Reporter { peer: context.peer.clone(), token: context.meta.get_progress_token(), last: 0.0 };
loop {
let state = self.job_state(job).await;
let finished = state.as_ref().is_none_or(|j| j["finished"].as_bool() == Some(true));
if let Some(j) = &state {
let (done, total) = (j["done"].as_f64().unwrap_or(0.0), j["total"].as_f64().unwrap_or(0.0));
rep.report(done, total, j["status"].as_str().unwrap_or_default()).await;
}
if finished {
let result = state.map(|j| j["result"].clone()).unwrap_or(Value::Null);
if let Some(e) = result.get("error").and_then(Value::as_str) {
return CallToolResult::error(vec![Content::text(format!("export failed: {e}"))]);
}
let mut out = start;
if let Some(o) = out.as_object_mut() {
o.insert("result".into(), result);
}
return CallToolResult::success(vec![Content::text(serde_json::to_string_pretty(&out).unwrap_or_default())]);
}
tokio::select! {
_ = context.ct.cancelled() => break,
_ = tokio::time::sleep(POLL) => {}
}
}
// Cancelled: stop the encode at the next batch, wait for the worker, delete the partial file.
let _ = self.run("jobs.cancel", json!({"job": job})).await;
for _ in 0..600 {
if self.job_state(job).await.is_none_or(|j| j["finished"].as_bool() == Some(true)) {
break;
}
tokio::time::sleep(POLL).await;
}
remove_partial(&path, started);
CallToolResult::error(vec![Content::text("cancelled")])
}
/// The job's `jobs.list` entry.
async fn job_state(&self, job: u64) -> Option<Value> {
let jobs = self.run("jobs.list", json!({})).await.ok()?;
jobs.as_array()?.iter().find(|j| j["id"].as_u64() == Some(job)).cloned()
}
}
/// Delete what an interrupted export wrote: the output file and, for image sequences and caption
/// sidecars, the files next to it named `<stem>[digits].<ext>` — only those written since `since`.
pub fn remove_partial(path: &str, since: SystemTime) {
let p = Path::new(path);
let (Some(dir), Some(stem)) = (p.parent(), p.file_stem().map(|s| s.to_string_lossy().to_string())) else { return };
let dir = if dir.as_os_str().is_empty() { Path::new(".") } else { dir };
let Ok(entries) = std::fs::read_dir(dir) else { return };
// file times are coarser than the clock
let since = since.checked_sub(Duration::from_secs(1)).unwrap_or(since);
for e in entries.flatten() {
let name = e.file_name().to_string_lossy().to_string();
let sibling = name.strip_prefix(&stem).is_some_and(|rest| rest.trim_start_matches(|c: char| c.is_ascii_digit()).starts_with('.'));
let ours = sibling && e.metadata().is_ok_and(|m| m.is_file() && m.modified().is_ok_and(|t| t >= since));
if ours {
let _ = std::fs::remove_file(e.path());
}
}
}

View File

@ -564,7 +564,7 @@ impl ServerHandler for FilmcraftMcp {
.with_instructions(INSTRUCTIONS.to_string())
}
/// Strict arguments in front of the tool router.
/// Strict arguments and long exports in front of the tool router.
async fn call_tool(
&self,
request: rmcp::model::CallToolRequestParams,
@ -573,6 +573,17 @@ impl ServerHandler for FilmcraftMcp {
if let Some(m) = self.unknown_arg(&request.name, request.arguments.as_ref()) {
return Err(McpError::invalid_params(m, None));
}
// A blocking export reports progress and can be cancelled (docs/agents.md § Long exports).
if request.name == "command_run"
&& matches!(&*self.backend, Backend::Headless(_))
&& let Some(a) = &request.arguments
&& let Some(id) = a.get("id").and_then(Value::as_str)
{
let params = a.get("params").cloned().filter(Value::is_object).unwrap_or_else(|| json!({}));
if crate::long_job::is_long(id, &params) {
return Ok(self.run_long(id, params, &context).await.into());
}
}
let tcc = rmcp::handler::server::tool::ToolCallContext::new(self, request, context);
self.tool_router.call(tcc).await
}
@ -784,6 +795,83 @@ mod tests {
}
}
/// docs/agents.md § Long exports: a blocking export reports `notifications/progress` for its
/// token (none without one), answers `ping` meanwhile, and `notifications/cancelled` stops
/// it, deletes the partial file and sends no response.
#[tokio::test(flavor = "multi_thread")]
async fn export_progress_and_cancel() {
let dir = std::env::temp_dir().join(format!("filmcraft-mcp-progress-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let mut c = Client::start(demo());
c.init().await;
let export = |id: u64, name: &str, token: Option<Value>, seconds: f64| {
let path = dir.join(name).to_string_lossy().to_string();
let params = json!({"path": path, "format": "h264", "width": 640, "height": 360, "audio": false, "range": "custom", "startSeconds": 0, "endSeconds": seconds, "wait": true});
let mut call =
json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"command_run","arguments":{"id":"file.exportMedia","params":params}}});
if let Some(t) = token {
call["params"]["_meta"] = json!({"progressToken": t});
}
call.to_string()
};
// no token: no notifications, the same response as before
c.send(&export(1, "quiet.mp4", None, 0.5)).await;
let m = c.next().await;
assert_eq!((m["id"].as_u64(), &m["result"]["isError"]), (Some(1), &json!(false)), "{m}");
// progress, strictly increasing, with a total; ping answered while the export runs
c.send(&export(2, "a.mp4", Some(json!("tok")), 1.0)).await;
let (mut last, mut notes, mut pinged, mut pong) = (-1.0, 0, false, false);
let done = loop {
let m = c.next().await;
if m["method"] == "notifications/progress" {
assert_eq!(m["params"]["progressToken"], "tok", "{m}");
let p = m["params"]["progress"].as_f64().unwrap();
assert!(p > last && m["params"]["total"].as_f64().unwrap() >= p, "{m}");
last = p;
notes += 1;
if !pinged {
pinged = true;
c.send(r#"{"jsonrpc":"2.0","id":3,"method":"ping"}"#).await;
}
} else if m["id"] == 3 {
pong = true;
} else if m["id"] == 2 {
break m;
}
};
assert_eq!(done["result"]["isError"], false, "{done}");
assert!(notes >= 2 && pong, "{notes} notifications, ping answered during the export: {pong}");
assert!(dir.join("a.mp4").exists());
// cancel after the first notification: no response, no partial file
c.send(&export(4, "b.mp4", Some(json!(7)), 8.0)).await;
loop {
let m = c.next().await;
assert_ne!(m["id"], 4, "finished before the cancel: {m}");
if m["method"] == "notifications/progress" {
break;
}
}
c.send(r#"{"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":4}}"#).await;
let t0 = std::time::Instant::now();
while dir.join("b.mp4").exists() || t0.elapsed() < std::time::Duration::from_millis(500) {
assert!(t0.elapsed().as_secs() < 30, "partial file still there");
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
c.send(r#"{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{"name":"command_run","arguments":{"id":"jobs.list"}}}"#).await;
let m = loop {
let m = c.next().await;
assert_ne!(m["id"], 4, "no response for a cancelled request: {m}");
if m["id"] == 5 {
break m;
}
};
let jobs: Value = serde_json::from_str(m["result"]["content"][0]["text"].as_str().unwrap()).unwrap();
let last = jobs.as_array().and_then(|a| a.last()).cloned().unwrap_or_default();
assert_eq!(last["result"]["error"], "cancelled", "{jobs}");
assert!(!dir.join("b.mp4").exists());
let _ = std::fs::remove_dir_all(&dir);
}
/// A command that panicked used to poison the session lock, and every later call failed with
/// "session lock poisoned" until the server was restarted.
#[tokio::test(flavor = "multi_thread")]

View File

@ -81,6 +81,24 @@ driven one app can drive the others.
2026-07-28 clients such as current Claude Code, list and read results carry the `ttlMs` and
`cacheScope` hints that revision requires.
### Long exports: progress and cancellation
`command_run {"id": "file.exportMedia", "params": {…, "wait": true}}` in headless mode blocks until
the file is written. It uses the MCP progress and cancellation utilities, so a client can show
progress and stop it:
- The export runs as the same background job `jobs.list` shows. The session is locked only for short
polls, so other requests (`ping`, `doc_inspect`, `jobs.list`, …) are answered while it encodes.
- With `"_meta": {"progressToken": T}` on the `tools/call`, the server sends
`notifications/progress` `{progressToken: T, progress: <frames done>, total: <frames>, message:
<job status>}` at most every 100 ms while frames advance. Without a token it sends none.
- `notifications/cancelled` for the request stops the encode at the next batch (`jobs.cancel`),
deletes the partial output (the movie, or the frames and caption sidecar written so far) and sends
no response, as the MCP cancellation utility asks.
- Without `wait` the command returns `{job, path}` at once, as before: poll `jobs.list` and stop it
with `jobs.cancel`. Other exports (`file.exportFrame`, interchange formats, …) finish quickly and
ignore the token. Bridge mode forwards the call unchanged; the app shows its own progress.
## 2. Control channel
`filmcraft --control 9876` (or `FILMCRAFT_CONTROL_PORT=9876`) listens on `127.0.0.1` only. Send one

View File

@ -106,6 +106,7 @@ or `--bridge 127.0.0.1:9876` to drive the running app. Tools: `command_list`, `c
`command_batch`, `doc_inspect`, `render_preview`, `project_inspect`, `sequence_inspect`, `media_import`,
`render_frame`, `ui_inspect`, `ui_elements`, `ui_click`, `ui_drag`, `ui_key`, `ui_type`, `ui_screenshot`,
`ui_control`; resources `filmcraft://document` and `filmcraft://commands`. `.mcp.json` registers both.
Annotations, argument checking and errors: [agents.md § Conventions](agents.md#conventions).
Annotations, argument checking, errors and export progress / cancellation:
[agents.md § Conventions](agents.md#conventions).
Time is in ticks (254 016 000 000 per second); commands also accept `seconds`, `frame` or `timecode`.