diff --git a/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json b/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json index 933d870b73..bed99ef1b7 100644 --- a/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json +++ b/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json @@ -43,8 +43,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json b/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json index aeb1091238..358758aff7 100644 --- a/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json +++ b/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json @@ -42,8 +42,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-50f81c128d644c60837e603099e169fbc7a500c11e79d2c78d6eccccf6491aec.json b/backend/.sqlx/query-50f81c128d644c60837e603099e169fbc7a500c11e79d2c78d6eccccf6491aec.json index 58bac5a002..63243f0b4f 100644 --- a/backend/.sqlx/query-50f81c128d644c60837e603099e169fbc7a500c11e79d2c78d6eccccf6491aec.json +++ b/backend/.sqlx/query-50f81c128d644c60837e603099e169fbc7a500c11e79d2c78d6eccccf6491aec.json @@ -38,8 +38,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json b/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json index 19b977d8da..9422e3e8d0 100644 --- a/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json +++ b/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json @@ -77,8 +77,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json b/backend/.sqlx/query-7aaa5b0bd873c2029e2201d287ea0aaae04678ac105374bbe387e534a6cb6333.json similarity index 74% rename from backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json rename to backend/.sqlx/query-7aaa5b0bd873c2029e2201d287ea0aaae04678ac105374bbe387e534a6cb6333.json index 3c24c19c8d..667a2d7b67 100644 --- a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json +++ b/backend/.sqlx/query-7aaa5b0bd873c2029e2201d287ea0aaae04678ac105374bbe387e534a6cb6333.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT runnable_id as \"runnable_id: ScriptHash\", raw_flow as \"raw_flow: _\", kind as \"kind: _\" FROM v2_job WHERE id = $1", + "query": "SELECT runnable_id as \"runnable_id: ScriptHash\", raw_flow as \"raw_flow: _\", kind as \"kind: _\", parent_job, flow_step_id FROM v2_job WHERE id = $1", "describe": { "columns": [ { @@ -42,12 +42,21 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } } + }, + { + "ordinal": 3, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 4, + "name": "flow_step_id", + "type_info": "Varchar" } ], "parameters": { @@ -58,8 +67,10 @@ "nullable": [ true, true, - false + false, + true, + true ] }, - "hash": "805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5" + "hash": "7aaa5b0bd873c2029e2201d287ea0aaae04678ac105374bbe387e534a6cb6333" } diff --git a/backend/.sqlx/query-7bb8ff3426ec73672fb572cd1ce36f495a84e0ad7d60fd39eddcccbc129d43c8.json b/backend/.sqlx/query-7bb8ff3426ec73672fb572cd1ce36f495a84e0ad7d60fd39eddcccbc129d43c8.json index 5d0d2f6938..66baa9f216 100644 --- a/backend/.sqlx/query-7bb8ff3426ec73672fb572cd1ce36f495a84e0ad7d60fd39eddcccbc129d43c8.json +++ b/backend/.sqlx/query-7bb8ff3426ec73672fb572cd1ce36f495a84e0ad7d60fd39eddcccbc129d43c8.json @@ -44,8 +44,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-8d444380516faf56369cdbff645904e20477aa597e30891926af42eee56ab081.json b/backend/.sqlx/query-8d444380516faf56369cdbff645904e20477aa597e30891926af42eee56ab081.json index 4840a3c28d..497bb3ac8f 100644 --- a/backend/.sqlx/query-8d444380516faf56369cdbff645904e20477aa597e30891926af42eee56ab081.json +++ b/backend/.sqlx/query-8d444380516faf56369cdbff645904e20477aa597e30891926af42eee56ab081.json @@ -102,8 +102,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json index 51dec42770..f39f1be51b 100644 --- a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json +++ b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json @@ -32,8 +32,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json b/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json index 9ba91640e8..33636da608 100644 --- a/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json +++ b/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json @@ -72,8 +72,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json index 43b1e07e2c..03ee58ca3f 100644 --- a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json +++ b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json @@ -77,8 +77,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-b23784ed2737c01aaf05efe42bcdcfc9579aa203005ead60322fb71fd33b15b0.json b/backend/.sqlx/query-b23784ed2737c01aaf05efe42bcdcfc9579aa203005ead60322fb71fd33b15b0.json index 36f9127807..054521168e 100644 --- a/backend/.sqlx/query-b23784ed2737c01aaf05efe42bcdcfc9579aa203005ead60322fb71fd33b15b0.json +++ b/backend/.sqlx/query-b23784ed2737c01aaf05efe42bcdcfc9579aa203005ead60322fb71fd33b15b0.json @@ -102,8 +102,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json index f06151fec9..4fa7da00e0 100644 --- a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json +++ b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json @@ -72,8 +72,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json index dde67a2606..4118760af2 100644 --- a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json +++ b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json @@ -41,8 +41,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json index 4160b79be6..9d2f8d9a9f 100644 --- a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json +++ b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json @@ -41,8 +41,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-e1f10f940d7ba6b1652e3d6505e12b16394f9632fd535febb008ebfe9fe0b7c8.json b/backend/.sqlx/query-e1f10f940d7ba6b1652e3d6505e12b16394f9632fd535febb008ebfe9fe0b7c8.json index d7837e7f79..18dbf7953e 100644 --- a/backend/.sqlx/query-e1f10f940d7ba6b1652e3d6505e12b16394f9632fd535febb008ebfe9fe0b7c8.json +++ b/backend/.sqlx/query-e1f10f940d7ba6b1652e3d6505e12b16394f9632fd535febb008ebfe9fe0b7c8.json @@ -31,8 +31,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json b/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json index ef7541aa28..cab697d254 100644 --- a/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json +++ b/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json @@ -37,8 +37,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json index e5bf2cbffc..7309b03a02 100644 --- a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json +++ b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json @@ -77,8 +77,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json b/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json index 3c8487cec3..d8855042a2 100644 --- a/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json +++ b/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json @@ -32,8 +32,7 @@ "aiagent", "unassigned_script", "unassigned_flow", - "unassigned_singlestepflow", - "snapshotbuild" + "unassigned_singlestepflow" ] } } diff --git a/backend/windmill-types/src/flows.rs b/backend/windmill-types/src/flows.rs index a1a2038452..c69090a547 100644 --- a/backend/windmill-types/src/flows.rs +++ b/backend/windmill-types/src/flows.rs @@ -540,53 +540,48 @@ impl FlowModule { ) -> anyhow::Result<()> { for module in modules { cb(module)?; - match module + let module_value = module .get_value() - .map_err(|e| anyhow::anyhow!("Module '{}': {}", module.id, e))? - { - FlowModuleValue::ForloopFlow { modules, .. } - | FlowModuleValue::WhileloopFlow { modules, .. } => { - Self::traverse_modules(&modules, cb)?; - } - FlowModuleValue::BranchOne { branches, default, .. } => { - for branch in branches { - Self::traverse_modules(&branch.modules, cb)?; - } - Self::traverse_modules(&default, cb)?; - } - FlowModuleValue::BranchAll { branches, .. } => { - for branch in branches { - Self::traverse_modules(&branch.modules, cb)?; - } - } - FlowModuleValue::AIAgent { tools, .. } => { - for tool in tools { - match &tool.value { - ToolValue::FlowModule(module_value) => match module_value { - FlowModuleValue::ForloopFlow { modules, .. } - | FlowModuleValue::WhileloopFlow { modules, .. } => { - Self::traverse_modules(&modules, cb)?; - } - FlowModuleValue::BranchOne { branches, default, .. } => { - for branch in branches { - Self::traverse_modules(&branch.modules, cb)?; - } - Self::traverse_modules(&default, cb)?; - } - FlowModuleValue::BranchAll { branches, .. } => { - for branch in branches { - Self::traverse_modules(&branch.modules, cb)?; - } - } - _ => {} - }, - ToolValue::Mcp(_) => {} - ToolValue::Websearch(_) => {} - } - } - } - _ => {} + .map_err(|e| anyhow::anyhow!("Module '{}': {}", module.id, e))?; + Self::traverse_module_value(&module_value, cb)?; + } + Ok(()) + } + + fn traverse_module_value anyhow::Result<()>>( + module_value: &FlowModuleValue, + cb: &mut C, + ) -> anyhow::Result<()> { + match module_value { + FlowModuleValue::ForloopFlow { modules, .. } + | FlowModuleValue::WhileloopFlow { modules, .. } => { + Self::traverse_modules(modules, cb)?; } + FlowModuleValue::BranchOne { branches, default, .. } => { + for branch in branches { + Self::traverse_modules(&branch.modules, cb)?; + } + Self::traverse_modules(default, cb)?; + } + FlowModuleValue::BranchAll { branches, .. } => { + for branch in branches { + Self::traverse_modules(&branch.modules, cb)?; + } + } + FlowModuleValue::AIAgent { tools, .. } => { + for tool in tools { + let Some(tool_module) = Option::::from(tool) else { + continue; + }; + + cb(&tool_module)?; + let tool_value = tool_module + .get_value() + .map_err(|e| anyhow::anyhow!("Tool module '{}': {}", tool_module.id, e))?; + Self::traverse_module_value(&tool_value, cb)?; + } + } + _ => {} } Ok(()) } @@ -1071,7 +1066,10 @@ impl Into> for FlowModuleValue { } } -pub fn ordered_map(value: &HashMap, serializer: S) -> Result +pub fn ordered_map( + value: &HashMap, + serializer: S, +) -> Result where S: Serializer, { diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 5db641b80d..a95f7bd4ab 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -3,8 +3,8 @@ use crate::ai::types::McpToolSource; use crate::ai::types::*; use crate::ai::utils::{ add_message_to_conversation, execute_mcp_tool, get_step_name_from_flow, - update_flow_status_module_with_actions, update_flow_status_module_with_actions_success, - FlowContext, + is_completed_input_transform, update_flow_status_module_with_actions, + update_flow_status_module_with_actions_success, FlowContext, }; use crate::common::OccupancyMetrics; use crate::result_processor::handle_non_flow_job_error; @@ -21,7 +21,6 @@ use serde_json::value::RawValue; use std::{collections::HashMap, sync::Arc}; use uuid::Uuid; use windmill_common::ai_types::OpenAIToolCall; -use windmill_common::flows::InputTransform; use windmill_common::jobs::JobPayload; #[cfg(feature = "mcp")] @@ -35,15 +34,15 @@ type McpClient = McpClientStub; use windmill_common::{ client::AuthedClient, db::DB, - error::{to_anyhow, Error}, + error::Error, flow_conversations::MessageType, flow_status::AgentAction, flows::FlowModuleValue, worker::{to_raw_value, Connection}, }; use windmill_queue::{ - get_mini_pulled_job, push, JobCompleted, MiniCompletedJob, MiniPulledJob, PushArgs, - PushIsolationLevel, + add_completed_job, add_completed_job_error, get_mini_pulled_job, push, MiniCompletedJob, + MiniPulledJob, PushArgs, PushIsolationLevel, }; /// Context for tool execution containing all required references and state @@ -54,8 +53,9 @@ pub struct ToolExecutionContext<'a> { // Job context pub job: &'a MiniPulledJob, - pub parent_job: &'a Uuid, + pub parent_job: Option<&'a Uuid>, pub summary: &'a Option<&'a str>, + pub flow_step_id_override: Option<&'a str>, // Execution parameters pub client: &'a AuthedClient, @@ -66,7 +66,6 @@ pub struct ToolExecutionContext<'a> { // Runtime state pub occupancy_metrics: &'a mut OccupancyMetrics, - pub job_completed_tx: &'a JobCompletedSender, pub killpill_rx: &'a mut tokio::sync::broadcast::Receiver<()>, // Optional streaming & chat @@ -283,7 +282,9 @@ async fn execute_windmill_tool( module_id: tool_module.id.clone(), }); - update_flow_status_module_with_actions(ctx.db, ctx.parent_job, actions).await?; + if let Some(parent_job) = ctx.parent_job { + update_flow_status_module_with_actions(ctx.db, parent_job, actions).await?; + } let raw_tool_call_args = if tool_call.function.arguments.is_empty() { "{}".to_string() @@ -301,11 +302,14 @@ async fn execute_windmill_tool( ) })?; + let tool_value = tool_module.get_value()?; + // Get input transforms given by the user and merge them with AI given args - let input_transforms = match tool_module.get_value()? { - FlowModuleValue::Script { input_transforms, .. } => input_transforms, - FlowModuleValue::RawScript { input_transforms, .. } => input_transforms, - FlowModuleValue::FlowScript { input_transforms, .. } => input_transforms, + let input_transforms = match &tool_value { + FlowModuleValue::Script { input_transforms, .. } + | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::FlowScript { input_transforms, .. } + | FlowModuleValue::AIAgent { input_transforms, .. } => input_transforms, _ => { return Err(Error::internal_err(format!( "Unsupported tool: {}", @@ -331,17 +335,8 @@ async fn execute_windmill_tool( // Evaluate each input transform and merge with AI-provided args for (key, transform) in input_transforms.iter() { // We skip static empty / null values, those are the one the AI will fill in - match transform { - InputTransform::Static { value } => { - let val = value.get().trim(); - if val.is_empty() || val == "null" { - continue; - } - } - InputTransform::Ai => { - continue; - } - _ => (), + if !is_completed_input_transform(transform) { + continue; } let result = evaluate_input_transform::>( transform, @@ -356,7 +351,7 @@ async fn execute_windmill_tool( tool_call_args.insert(key.clone(), result); } - let job_payload = match tool_module.get_value()? { + let job_payload = match tool_value { FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => { script_to_payload( script_hash, @@ -380,7 +375,6 @@ async fn execute_windmill_tool( } => { let path = path .unwrap_or_else(|| format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id)); - raw_script_to_payload( path, content, @@ -394,8 +388,7 @@ async fn execute_windmill_tool( } FlowModuleValue::FlowScript { id, language, concurrency_settings, tag, .. } => { let path = format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id); - - let payload = JobPayloadWithTag { + JobPayloadWithTag { payload: JobPayload::FlowScript { id, language, @@ -409,8 +402,29 @@ async fn execute_windmill_tool( delete_after_use: tool_module.delete_after_use.unwrap_or(false), timeout: None, on_behalf_of: None, - }; - payload + } + } + FlowModuleValue::AIAgent { tools: sub_tools, .. } => { + let has_nested_agent_tools = sub_tools.iter().any(|t| { + matches!( + t.value, + windmill_common::flows::ToolValue::FlowModule(FlowModuleValue::AIAgent { .. }) + ) + }); + if has_nested_agent_tools { + return Err(Error::internal_err( + "AI agent tools cannot be nested beyond 2 levels. The nested agent tool contains \ + AIAgent sub-tools, which would exceed the maximum nesting depth.".to_string() + )); + } + let path = format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id); + JobPayloadWithTag { + payload: JobPayload::AIAgent { path }, + tag: None, + delete_after_use: tool_module.delete_after_use.unwrap_or(false), + timeout: None, + on_behalf_of: None, + } } _ => { return Err(Error::internal_err(format!( @@ -452,8 +466,8 @@ async fn execute_windmill_tool( None, ctx.job.schedule_path(), Some(ctx.job.id), - None, - None, + ctx.job.root_job.or(Some(ctx.job.id)), + ctx.job.flow_innermost_root_job.or(Some(ctx.job.id)), Some(job_id), false, false, @@ -544,7 +558,6 @@ async fn execute_windmill_tool( ctx.occupancy_metrics.total_duration_of_running_jobs = updated_occupancy.total_duration_of_running_jobs; - // Continue with match on handle_result match handle_result { Err(err) => { handle_tool_execution_error( @@ -627,7 +640,9 @@ async fn handle_tool_execution_error( .await?; } - update_flow_status_module_with_actions_success(ctx.db, ctx.parent_job, false).await?; + if let Some(parent_job) = ctx.parent_job { + update_flow_status_module_with_actions_success(ctx.db, parent_job, false).await?; + } // Add tool message to conversation if chat_input_enabled (error case) add_tool_message_to_chat(ctx, Some(job_id), &error_message, false).await; @@ -649,23 +664,50 @@ async fn handle_tool_execution_success( let send_result = inner_job_completed_rx.bounded_rx.try_recv().ok(); let result = if let Some(SendResult { - result: SendResultPayload::JobCompleted(JobCompleted { result, .. }), - .. - }) = send_result.as_ref() + result: SendResultPayload::JobCompleted(ref jc), .. + }) = send_result { - let result = result.clone(); - ctx.job_completed_tx - .send(send_result.unwrap().result, true) + let result = jc.result.clone(); + // Write tool completion to the DB inline instead of forwarding through + // the parent channel. Forwarding would deadlock for nested agents: the + // sub-tool result would fill the parent's bounded(1) channel, leaving + // no room for the agent's own completion from process_result. + if jc.success { + add_completed_job( + ctx.db, + &jc.job, + true, + false, + sqlx::types::Json(&*jc.result), + jc.result_columns.clone(), + jc.mem_peak, + jc.canceled_by.clone(), + false, + jc.duration, + jc.from_cache.unwrap_or(false), + ) .await - .map_err(to_anyhow)?; + .map_err(|e| Error::internal_err(format!("Failed to add completed job: {e}")))?; + } else { + let error_value: serde_json::Value = + serde_json::from_str(jc.result.get()).unwrap_or_else(|_| { + serde_json::json!({ "message": format!("Non serializable error: {}", jc.result.get()) }) + }); + add_completed_job_error( + ctx.db, + &jc.job, + jc.mem_peak, + jc.canceled_by.clone(), + error_value, + ctx.worker_name, + false, + jc.duration, + ) + .await + .map_err(|e| Error::internal_err(format!("Failed to add completed job error: {e}")))?; + } result } else { - if let Some(send_result) = send_result { - ctx.job_completed_tx - .send(send_result.result, true) - .await - .map_err(to_anyhow)?; - } return Err(Error::internal_err( "Tool job completed but no result".to_string(), )); @@ -696,7 +738,9 @@ async fn handle_tool_execution_success( .await?; } - update_flow_status_module_with_actions_success(ctx.db, ctx.parent_job, success).await?; + if let Some(parent_job) = ctx.parent_job { + update_flow_status_module_with_actions_success(ctx.db, parent_job, success).await?; + } // Add tool message to conversation if chat_input_enabled let content = if success { @@ -731,8 +775,10 @@ async fn add_tool_message_to_chat( .and_then(|fs| fs.memory_id) { let db_clone = ctx.db.clone(); - let step_name = - get_step_name_from_flow(ctx.summary.as_deref(), ctx.job.flow_step_id.as_deref()); + let effective_step_id = ctx + .flow_step_id_override + .or(ctx.job.flow_step_id.as_deref()); + let step_name = get_step_name_from_flow(ctx.summary.as_deref(), effective_step_id); let content = content.to_string(); // Spawn task because we do not need to wait for the result diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index 7bd02af4ff..a374db29c5 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -62,6 +62,17 @@ pub fn parse_raw_script_schema( Ok(to_raw_value(&schema)) } +pub fn is_completed_input_transform(transform: &InputTransform) -> bool { + match transform { + InputTransform::Static { value } => { + let val = value.get().trim(); + !val.is_empty() && val != "null" + } + InputTransform::Javascript { expr } => !expr.trim().is_empty(), + InputTransform::Ai => false, + } +} + /// Filters out properties from a JSON schema that have completed input transforms. /// This allows AI agents to only see and fill parameters that don't have user-configured values. pub fn filter_schema_by_input_transforms( @@ -77,14 +88,7 @@ pub fn filter_schema_by_input_transforms( let keys_to_remove: HashSet = input_transforms .iter() .filter_map(|(key, transform)| { - let is_completed = match transform { - InputTransform::Static { value } => { - let val = value.get().trim(); - !val.is_empty() && val != "null" - } - InputTransform::Javascript { expr } => !expr.trim().is_empty(), - InputTransform::Ai => false, - }; + let is_completed = is_completed_input_transform(transform); if is_completed { Some(key.clone()) } else { @@ -123,10 +127,13 @@ pub fn filter_schema_by_input_transforms( Ok(to_raw_value(&schema_value)) } +#[derive(Clone)] pub struct FlowJobRunnableIdAndRawFlow { pub runnable_id: Option, pub raw_flow: Option>>, pub kind: JobKind, + pub parent_job: Option, + pub flow_step_id: Option, } pub async fn get_flow_job_runnable_and_raw_flow( @@ -135,7 +142,7 @@ pub async fn get_flow_job_runnable_and_raw_flow( ) -> windmill_common::error::Result { let job = sqlx::query_as!( FlowJobRunnableIdAndRawFlow, - "SELECT runnable_id as \"runnable_id: ScriptHash\", raw_flow as \"raw_flow: _\", kind as \"kind: _\" FROM v2_job WHERE id = $1", + "SELECT runnable_id as \"runnable_id: ScriptHash\", raw_flow as \"raw_flow: _\", kind as \"kind: _\", parent_job, flow_step_id FROM v2_job WHERE id = $1", job_id ) .fetch_one(db) @@ -690,6 +697,7 @@ pub fn any_tool_needs_previous_result(tools: &[Tool]) -> bool { FlowModuleValue::Script { input_transforms, .. } => input_transforms, FlowModuleValue::RawScript { input_transforms, .. } => input_transforms, FlowModuleValue::FlowScript { input_transforms, .. } => input_transforms, + FlowModuleValue::AIAgent { input_transforms, .. } => input_transforms, _ => return false, }; diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 3a721b4a5e..e0271ecb8a 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -47,7 +47,6 @@ use crate::{ }, common::{build_args_map, resolve_job_timeout, OccupancyMetrics, StreamNotifier}, handle_child::run_future_with_polling_update_job_poller, - JobCompletedSender, }; lazy_static::lazy_static! { @@ -79,11 +78,57 @@ lazy_static::lazy_static! { }) .unwrap_or_default() }; + + static ref AI_AGENT_TOOL_SCHEMA: Box = to_raw_value(&serde_json::json!({ + "type": "object", + "properties": { + "user_message": { "type": "string" }, + }, + "required": ["user_message"], + "additionalProperties": false, + })); } const DEFAULT_MAX_AGENT_ITERATIONS: usize = 10; const HARD_MAX_AGENT_ITERATIONS: usize = 1000; +fn find_module_by_id( + modules: &Vec, + target_id: &str, +) -> Result, Error> { + let mut found: Option = None; + FlowModule::traverse_modules(modules, &mut |module| { + if found.is_none() && module.id == target_id { + found = Some(module.clone()); + } + Ok(()) + }) + .map_err(|e| Error::internal_err(format!("Failed to traverse flow modules: {e}")))?; + Ok(found) +} + +fn find_ai_agent_tool_module_in_parent_agent( + modules: &Vec, + parent_agent_step_id: &str, + tool_module_id: &str, +) -> Result, Error> { + let Some(parent_agent_module) = find_module_by_id(modules, parent_agent_step_id)? else { + return Ok(None); + }; + + let FlowModuleValue::AIAgent { tools, .. } = parent_agent_module.get_value()? else { + return Ok(None); + }; + + for tool in tools { + if tool.id == tool_module_id { + return Ok(Option::::from(&tool)); + } + } + + Ok(None) +} + pub async fn handle_ai_agent_job( // connection conn: &Connection, @@ -97,7 +142,6 @@ pub async fn handle_ai_agent_job( canceled_by: &mut Option, mem_peak: &mut i32, occupancy_metrics: &mut OccupancyMetrics, - job_completed_tx: &JobCompletedSender, worker_dir: &str, base_internal_url: &str, worker_name: &str, @@ -117,26 +161,57 @@ pub async fn handle_ai_agent_job( return handle_credentials_check(&args.provider).await; } - let Some(flow_step_id) = &job.flow_step_id else { - return Err(Error::internal_err( - "AI agent job has no flow step id".to_string(), - )); - }; + // flow_step_id is set by the flow executor for top-level AI agents. + // For nested AI agent tools, it's not set (to avoid triggering flow step + // machinery on a parent that has no v2_job_status row), so we extract the + // tool module ID from the runnable_path which has the form ".../tools/{id}". + let flow_step_id = job + .flow_step_id + .as_deref() + .or_else(|| job.runnable_path().rsplit_once("/tools/").map(|(_, id)| id)) + .ok_or_else(|| Error::internal_err("AI agent job has no flow step id".to_string()))? + .to_string(); + let flow_step_id = &flow_step_id; - let Some(parent_job) = &job.parent_job else { + let Some(immediate_parent_job) = &job.parent_job else { return Err(Error::internal_err( "AI agent job has no parent job".to_string(), )); }; - let flow_job = get_flow_job_runnable_and_raw_flow(db, &parent_job).await?; + let mut flow_job_id = *immediate_parent_job; + let mut flow_job = get_flow_job_runnable_and_raw_flow(db, &flow_job_id).await?; + let direct_parent_job_kind = flow_job.kind; + let direct_parent_job_flow_step_id = flow_job.flow_step_id.clone(); + + // If the direct parent is an AI agent (nested tool case), go one level up to the flow. + if flow_job.kind == JobKind::AIAgent { + let Some(parent_job_id) = flow_job.parent_job else { + return Err(Error::internal_err( + "AI agent parent has no parent job".to_string(), + )); + }; + flow_job_id = parent_job_id; + flow_job = get_flow_job_runnable_and_raw_flow(db, &flow_job_id).await?; + + if !matches!( + flow_job.kind, + JobKind::Flow | JobKind::FlowNode | JobKind::FlowPreview + ) { + return Err(Error::internal_err( + "AI agent nesting beyond 2 levels is not supported. \ + Only flow → agent → nested agent tool is allowed." + .to_string(), + )); + } + } let flow_data = match flow_job.kind { JobKind::Flow | JobKind::FlowNode => { cache::job::fetch_flow(db, &flow_job.kind, flow_job.runnable_id).await? } JobKind::FlowPreview => { - cache::job::fetch_preview_flow(db, &parent_job, flow_job.raw_flow).await? + cache::job::fetch_preview_flow(db, &flow_job_id, flow_job.raw_flow).await? } _ => { return Err(Error::internal_err( @@ -147,8 +222,18 @@ pub async fn handle_ai_agent_job( let value = flow_data.value(); - let module = value.modules.iter().find(|m| m.id == *flow_step_id); - let summary = module.as_ref().and_then(|m| m.summary.clone()); + let module = if direct_parent_job_kind == JobKind::AIAgent { + let parent_agent_step_id = direct_parent_job_flow_step_id.as_deref().ok_or_else(|| { + Error::internal_err("Parent AI agent job has no flow_step_id".to_string()) + })?; + find_ai_agent_tool_module_in_parent_agent( + &value.modules, + parent_agent_step_id, + flow_step_id, + )? + } else { + find_module_by_id(&value.modules, flow_step_id)? + }; let Some(module) = module else { return Err(Error::internal_err( @@ -156,6 +241,8 @@ pub async fn handle_ai_agent_job( )); }; + let summary = module.summary.clone(); + let FlowModuleValue::AIAgent { tools, .. } = module.get_value()? else { return Err(Error::internal_err( "AI agent module is not an AI agent".to_string(), @@ -285,6 +372,16 @@ pub async fn handle_ai_agent_job( let schema = Some(parse_raw_script_schema(&content, &language)?); (schema, input_transforms) } + FlowModuleValue::AIAgent { input_transforms, .. } => { + // By convention for AIAgent tools, only user_message is expected to be AI-filled. + ( + Some( + RawValue::from_string(AI_AGENT_TOOL_SCHEMA.get().to_string()) + .expect("AI_AGENT_TOOL_SCHEMA should always be valid JSON"), + ), + input_transforms, + ) + } _ => { return Err(Error::internal_err(format!( "Unsupported tool: {}", @@ -342,18 +439,24 @@ pub async fn handle_ai_agent_job( stream_notifier.update_flow_status_with_stream_job(); } + let flow_status_job = if direct_parent_job_kind == JobKind::AIAgent { + None + } else { + Some(flow_job_id) + }; + let agent_fut = run_agent( db, conn, job, - parent_job, + flow_status_job.as_ref(), + Some(flow_step_id.as_str()), &args, &tools, &mcp_clients, summary.as_deref(), client, &mut inner_occupancy_metrics, - job_completed_tx, worker_dir, base_internal_url, worker_name, @@ -391,7 +494,8 @@ pub async fn run_agent( // agent job and flow data job: &MiniPulledJob, - parent_job: &Uuid, + parent_job: Option<&Uuid>, + flow_step_id_override: Option<&str>, args: &AIAgentArgs, tools: &[Tool], mcp_clients: &HashMap>, @@ -400,7 +504,6 @@ pub async fn run_agent( // job execution context client: &AuthedClient, occupancy_metrics: &mut OccupancyMetrics, - job_completed_tx: &JobCompletedSender, worker_dir: &str, base_internal_url: &str, worker_name: &str, @@ -433,6 +536,10 @@ pub async fn run_agent( vec![] }; + // Effective flow_step_id: override for nested agents, otherwise from job + let effective_flow_step_id: Option<&str> = + flow_step_id_override.or(job.flow_step_id.as_deref()); + // Fetch flow context for input transforms context, chat and memory let mut flow_context = get_flow_context(db, job).await; @@ -479,7 +586,7 @@ pub async fn run_agent( } Some(Memory::Auto { context_length, .. }) => { // Auto mode: load from memory - if let Some(step_id) = job.flow_step_id.as_deref() { + if let Some(step_id) = effective_flow_step_id { if let Some(memory_id) = memory_id { // Read messages from memory match read_from_memory(db, &job.workspace_id, memory_id, step_id).await { @@ -534,9 +641,8 @@ pub async fn run_agent( let id_context = { if let Some(ref flow_status) = flow_context.flow_status { // Get the step ID from the AI agent's flow step - let previous_id = job - .flow_step_id - .clone() + let previous_id = effective_flow_step_id + .map(str::to_string) .unwrap_or_else(|| "unknown".to_string()); Some(get_transform_context(job, &previous_id, flow_status)) @@ -649,7 +755,7 @@ pub async fn run_agent( .and_then(|fs| fs.chat_input_enabled) .unwrap_or(false); - let step_name = get_step_name_from_flow(summary.as_deref(), job.flow_step_id.as_deref()); + let step_name = get_step_name_from_flow(summary.as_deref(), effective_flow_step_id); let max_iterations = args .max_iterations @@ -881,8 +987,11 @@ pub async fn run_agent( ..Default::default() }); - update_flow_status_module_with_actions(db, parent_job, &actions).await?; - update_flow_status_module_with_actions_success(db, parent_job, true).await?; + if let Some(parent_job) = parent_job { + update_flow_status_module_with_actions(db, parent_job, &actions).await?; + update_flow_status_module_with_actions_success(db, parent_job, true) + .await?; + } content = Some(OpenAIContent::Text(response_content.clone())); @@ -940,13 +1049,13 @@ pub async fn run_agent( job, parent_job, summary: &summary, + flow_step_id_override, client, worker_dir, base_internal_url, worker_name, hostname, occupancy_metrics, - job_completed_tx, killpill_rx, stream_event_processor: stream_event_processor.as_ref(), flow_context: &mut flow_context, @@ -1071,7 +1180,7 @@ pub async fn run_agent( // final_messages contains the complete history (old messages + new ones) if matches!(output_type, OutputType::Text) && !use_manual_messages { if let Some(Memory::Auto { context_length, .. }) = &args.memory { - if let Some(step_id) = job.flow_step_id.as_deref() { + if let Some(step_id) = effective_flow_step_id { // Extract OpenAIMessages from final_messages let all_messages: Vec = final_messages.iter().map(|m| m.message.clone()).collect(); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 71d9f12a90..6f14512b32 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3398,7 +3398,6 @@ pub async fn handle_queued_job( &mut canceled_by, &mut mem_peak, &mut *occupancy_metrics, - &job_completed_tx, worker_dir, base_internal_url, worker_name, diff --git a/frontend/src/lib/components/InputTransformForm.svelte b/frontend/src/lib/components/InputTransformForm.svelte index 7f6352241f..98bb3035de 100644 --- a/frontend/src/lib/components/InputTransformForm.svelte +++ b/frontend/src/lib/components/InputTransformForm.svelte @@ -65,6 +65,7 @@ otherArgs?: Record helperScript?: DynamicInputTypes.HelperScript | undefined isAgentTool?: boolean + allowedAiTransforms?: string[] | undefined s3StorageConfigured?: boolean chatInputEnabled?: boolean } @@ -92,6 +93,7 @@ otherArgs = {}, helperScript = undefined, isAgentTool = false, + allowedAiTransforms = isAgentTool ? undefined : [], s3StorageConfigured = true, chatInputEnabled = false }: Props = $props() @@ -135,6 +137,11 @@ ) ) + // Whether this specific field is allowed to use AI transforms + let fieldAllowsAi = $derived( + allowedAiTransforms === undefined || allowedAiTransforms.includes(argName) + ) + let propertyType = $state(getPropertyType(arg)) function setExpr() { @@ -167,7 +174,7 @@ function getPropertyType(arg: InputTransform | any): InputTransform['type'] { // For agent tools, if static with undefined/empty value, treat as 'ai', meaning the field will be filled by the AI agent dynamically. if ( - isAgentTool && + fieldAllowsAi && ((arg?.type === 'static' && arg?.value === undefined) || arg?.type === 'ai') ) { if (arg?.type === 'static') { @@ -645,7 +652,7 @@ }} > {#snippet children({ item })} - {#if isAgentTool} + {#if fieldAllowsAi} - - + + This field will be filled by the AI agent dynamically diff --git a/frontend/src/lib/components/InputTransformSchemaForm.svelte b/frontend/src/lib/components/InputTransformSchemaForm.svelte index afb86a5a9e..20dd353b99 100644 --- a/frontend/src/lib/components/InputTransformSchemaForm.svelte +++ b/frontend/src/lib/components/InputTransformSchemaForm.svelte @@ -26,6 +26,7 @@ class?: string helperScript?: DynamicInputTypes.HelperScript isAgentTool?: boolean + allowedAiTransforms?: string[] | undefined chatInputEnabled?: boolean } @@ -42,6 +43,7 @@ class: clazz = '', helperScript = undefined, isAgentTool = false, + allowedAiTransforms = isAgentTool ? undefined : [], chatInputEnabled = false }: Props = $props() @@ -141,6 +143,7 @@ {enableAi} {helperScript} {isAgentTool} + {allowedAiTransforms} {s3StorageConfigured} {chatInputEnabled} otherArgs={Object.fromEntries( diff --git a/frontend/src/lib/components/flows/agentToolUtils.ts b/frontend/src/lib/components/flows/agentToolUtils.ts index ad9d1c0c7a..84e102ce56 100644 --- a/frontend/src/lib/components/flows/agentToolUtils.ts +++ b/frontend/src/lib/components/flows/agentToolUtils.ts @@ -1,8 +1,16 @@ -import type { AiAgent, FlowModule, FlowModuleValue } from '$lib/gen' +import type { AiAgent, FlowModule, FlowModuleValue, InputTransform } from '$lib/gen' +import { loadStoredConfig } from '../aiProviderStorage' +import { AI_AGENT_SCHEMA } from './flowInfers' + +export const SPECIAL_TOOL_KINDS = ['mcpTool', 'websearchTool', 'aiAgentTool'] as const +export type SpecialToolKind = (typeof SPECIAL_TOOL_KINDS)[number] // Type aliases for better readability export type AgentTool = AiAgent['tools'][number] export type FlowModuleTool = AgentTool & { value: { tool_type: 'flowmodule' } & FlowModuleValue } +export type AiAgentTool = AgentTool & { + value: { tool_type: 'flowmodule' } & { type: 'aiagent' } & FlowModuleValue +} export type McpTool = AgentTool & { value: { tool_type: 'mcp' @@ -38,6 +46,39 @@ export function isWebsearchTool(tool: AgentTool): tool is WebsearchTool { return tool.value.tool_type === 'websearch' } +/** + * Create an AI Agent tool (nested agent) + */ +export function createAiAgentTool(id: string): AiAgentTool { + const input_transforms: AiAgent['input_transforms'] = { + provider: { + type: 'static', + value: loadStoredConfig() ?? { kind: 'openai', resource: '', model: '' } + }, + output_type: { type: 'static', value: 'text' }, + user_message: { type: 'ai' } + } + for (const key of Object.keys(AI_AGENT_SCHEMA.properties ?? {})) { + if (!(key in input_transforms)) { + ;(input_transforms as Record)[key] = { + type: 'static', + value: undefined + } + } + } + + return { + id, + summary: '', + value: { + tool_type: 'flowmodule', + type: 'aiagent', + tools: [], + input_transforms + } + } as AiAgentTool +} + /** * Create an MCP tool from resource path */ @@ -67,6 +108,19 @@ export function createWebsearchTool(id: string): WebsearchTool { } } +/** + * Convert a FlowModuleTool to a FlowModule for use with loadFlowModuleState etc. + * Strips the extra `tool_type` field and maps AgentTool fields to FlowModule fields. + */ +export function agentToolToFlowModule(tool: FlowModuleTool): FlowModule { + const { tool_type: _, ...value } = tool.value + return { + id: tool.id, + summary: tool.summary, + value: value as FlowModuleValue + } +} + /** * Convert a FlowModule back to an AgentTool * Used when saving changes back to the AI Agent tools array diff --git a/frontend/src/lib/components/flows/content/AgentToolWrapper.svelte b/frontend/src/lib/components/flows/content/AgentToolWrapper.svelte index f61ea2141f..4d9b003d81 100644 --- a/frontend/src/lib/components/flows/content/AgentToolWrapper.svelte +++ b/frontend/src/lib/components/flows/content/AgentToolWrapper.svelte @@ -1,6 +1,11 @@