fix: notify on flow restore and resource changes, address review nits

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-09-24 09:50:51 +02:00
co-authored by Claude Opus 5.5
parent 51e0387d89
commit 6e071445f2
5 changed files with 46 additions and 16 deletions
@@ -1,6 +1,9 @@
DROP TRIGGER IF EXISTS script_mcp_tools_update_trigger ON script;
DROP TRIGGER IF EXISTS script_mcp_tools_delete_trigger ON script;
DROP TRIGGER IF EXISTS flow_mcp_tools_insert_trigger ON flow;
DROP TRIGGER IF EXISTS flow_mcp_tools_update_trigger ON flow;
DROP TRIGGER IF EXISTS resource_mcp_tools_insert_delete_trigger ON resource;
DROP TRIGGER IF EXISTS resource_mcp_tools_rename_trigger ON resource;
DROP TRIGGER IF EXISTS flow_mcp_tools_delete_trigger ON flow;
DROP TRIGGER IF EXISTS favorite_mcp_tools_insert_trigger ON favorite;
DROP TRIGGER IF EXISTS favorite_mcp_tools_delete_trigger ON favorite;
@@ -1,5 +1,5 @@
-- Deploys already emit `notify_runnable_version_change`; these cover the other changes to the
-- scripts and flows the MCP server lists as tools. One event per workspace per transaction:
-- Deploys already emit `notify_runnable_version_change`; these cover the other changes to what
-- the MCP server lists as tools. One event per workspace per transaction:
-- archiving a path touches every version of it and deleting a workspace every script, while the
-- poll loop reads 1000 events per tick, so a row-level burst would delay every other channel.
CREATE OR REPLACE FUNCTION notify_mcp_tools_change()
@@ -27,6 +27,14 @@ AFTER DELETE ON script
FOR EACH ROW
EXECUTE FUNCTION notify_mcp_tools_change();
-- A create inserts the row with no versions and appends one later, which the versions trigger
-- sees. Restore from trash, rename and fork insert the row with its versions already set.
CREATE TRIGGER flow_mcp_tools_insert_trigger
AFTER INSERT ON flow
FOR EACH ROW
WHEN (cardinality(NEW.versions) > 0)
EXECUTE FUNCTION notify_mcp_tools_change();
CREATE TRIGGER flow_mcp_tools_update_trigger
AFTER UPDATE OF archived ON flow
FOR EACH ROW
@@ -38,6 +46,18 @@ AFTER DELETE ON flow
FOR EACH ROW
EXECUTE FUNCTION notify_mcp_tools_change();
-- A tool's input schema lists the workspace's resources of each resource type it takes.
CREATE TRIGGER resource_mcp_tools_insert_delete_trigger
AFTER INSERT OR DELETE ON resource
FOR EACH ROW
EXECUTE FUNCTION notify_mcp_tools_change();
CREATE TRIGGER resource_mcp_tools_rename_trigger
AFTER UPDATE OF path ON resource
FOR EACH ROW
WHEN (OLD.path IS DISTINCT FROM NEW.path)
EXECUTE FUNCTION notify_mcp_tools_change();
-- Favorites are the tool list of an `mcp:favorites` token.
CREATE TRIGGER favorite_mcp_tools_insert_trigger
AFTER INSERT ON favorite
+4 -2
View File
@@ -1876,8 +1876,10 @@ async fn process_notify_event(
);
windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE.remove(payload);
}
#[cfg(feature = "mcp")]
"notify_mcp_tools_change" => windmill_api::notify_mcp_tools_changed(payload),
"notify_mcp_tools_change" => {
#[cfg(feature = "mcp")]
windmill_api::notify_mcp_tools_changed(payload);
}
"notify_runnable_version_change" => {
tracing::info!("Runnable version change detected: {}", payload);
match payload.split(':').collect::<Vec<&str>>().as_slice() {
+2 -2
View File
@@ -77,8 +77,6 @@ mod ai_shared_artifacts;
mod apps;
mod apps_raw_bundle;
pub use apps::invalidate_app_policy_cache;
#[cfg(feature = "mcp")]
pub use windmill_mcp::server::notify_tools_changed as notify_mcp_tools_changed;
pub mod args;
mod audit;
pub mod auth;
@@ -211,6 +209,8 @@ mod mcp_tools;
#[cfg(feature = "mcp")]
mod mcp;
#[cfg(feature = "mcp")]
pub use windmill_mcp::server::notify_tools_changed as notify_mcp_tools_changed;
#[cfg(all(feature = "mcp", feature = "private"))]
mod mcp_oauth_ee;
#[cfg(feature = "mcp")]
+15 -10
View File
@@ -57,16 +57,19 @@ const LIST_TTL_MS: u64 = 0;
/// cache entry would leak one token's view to another.
const LIST_CACHE_SCOPE: CacheScope = CacheScope::Private;
/// Workspaces whose listed scripts or flows changed. Fed from the notify-event
/// poll loop, so a change committed through any replica reaches the listen
/// streams held by every replica.
/// Workspaces whose tool listing changed. Fed from the notify-event poll loop, so
/// a change committed through any replica reaches the listen streams held by
/// every replica. Sized well past the poll loop's 1000-event batch: a lagged
/// receiver notifies, and every stream on the instance would re-list at once.
static TOOLS_CHANGED: LazyLock<broadcast::Sender<String>> =
LazyLock::new(|| broadcast::channel(1024).0);
LazyLock::new(|| broadcast::channel(8192).0);
/// A poll batch delivers all of its events back to back; waiting this long before
/// notifying folds a bulk change into one `list_changed` per stream.
const TOOLS_CHANGED_COALESCE: Duration = Duration::from_secs(1);
/// Wake the `subscriptions/listen` streams of `workspace_id` on this process.
/// Fire-and-forget: with no stream open, the event is dropped.
pub fn notify_tools_changed(workspace_id: &str) {
let _ = TOOLS_CHANGED.send(workspace_id.to_string());
}
@@ -465,10 +468,10 @@ impl<B: McpBackend> ServerHandler for Runner<B> {
Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS)
}
/// Only the pre-2026-07-28 revisions reach `initialize`, and they have no stream
/// to receive `list_changed` on (legacy sessions are off and
/// `subscriptions/listen` is 2026-07-28 only), so the capability is withdrawn
/// here. It stays in `get_info`, which bounds what a listen filter may accept.
/// A pre-2026-07-28 revision has no stream to receive `list_changed` on (legacy
/// sessions are off and `subscriptions/listen` is 2026-07-28 only), so the
/// capability is withdrawn when one is negotiated. It stays in `get_info`, which
/// bounds what a listen filter may accept.
async fn initialize(
&self,
request: InitializeRequestParams,
@@ -479,8 +482,10 @@ impl<B: McpBackend> ServerHandler for Runner<B> {
if SUPPORTED_PROTOCOL_VERSIONS.contains(&request.protocol_version) {
info.protocol_version = request.protocol_version;
}
if let Some(tools) = info.capabilities.tools.as_mut() {
tools.list_changed = None;
if info.protocol_version != ProtocolVersion::V_2026_07_28 {
if let Some(tools) = info.capabilities.tools.as_mut() {
tools.list_changed = None;
}
}
Ok(info)
}