diff --git a/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json b/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json new file mode 100644 index 0000000000..8fd07f3bf9 --- /dev/null +++ b/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json @@ -0,0 +1,77 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n v2_job.permissioned_as_email,\n v2_job.created_by,\n v2_job.parent_job,\n v2_job.permissioned_as,\n v2_job.runnable_path,\n CASE WHEN v2_job.trigger_kind = 'schedule'::job_trigger_kind THEN v2_job.trigger END AS schedule_path,\n v2_job.flow_step_id,\n v2_job.flow_innermost_root_job,\n v2_job.root_job,\n v2_job_queue.scheduled_for AS \"scheduled_for: chrono::DateTime\"\n FROM v2_job INNER JOIN v2_job_queue ON v2_job.id = v2_job_queue.id\n WHERE v2_job.id = $1 AND v2_job.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "permissioned_as_email", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "permissioned_as", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "schedule_path", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "flow_step_id", + "type_info": "Varchar" + }, + { + "ordinal": 7, + "name": "flow_innermost_root_job", + "type_info": "Uuid" + }, + { + "ordinal": 8, + "name": "root_job", + "type_info": "Uuid" + }, + { + "ordinal": 9, + "name": "scheduled_for: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + true, + false, + true, + null, + true, + true, + true, + false + ] + }, + "hash": "02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0" +} diff --git a/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json b/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json new file mode 100644 index 0000000000..34208d501c --- /dev/null +++ b/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as \"root_job!\" FROM v2_job WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "root_job!", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73" +} diff --git a/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json b/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json deleted file mode 100644 index fabcec9b2a..0000000000 --- a/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json +++ /dev/null @@ -1,71 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT\n email AS \"email!\",\n created_by AS \"created_by!\",\n parent_job, permissioned_as AS \"permissioned_as!\",\n script_path, schedule_path, flow_step_id, root_job,\n scheduled_for AS \"scheduled_for!: chrono::DateTime\"\n FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "email!", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "created_by!", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "parent_job", - "type_info": "Uuid" - }, - { - "ordinal": 3, - "name": "permissioned_as!", - "type_info": "Varchar" - }, - { - "ordinal": 4, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 5, - "name": "schedule_path", - "type_info": "Varchar" - }, - { - "ordinal": 6, - "name": "flow_step_id", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "root_job", - "type_info": "Uuid" - }, - { - "ordinal": 8, - "name": "scheduled_for!: chrono::DateTime", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - }, - "nullable": [ - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2" -} diff --git a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json similarity index 90% rename from backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json rename to backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json index fa33c3b8e0..28c0756b0f 100644 --- a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json +++ b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1", + "query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1", "describe": { "columns": [ { @@ -173,41 +173,46 @@ }, { "ordinal": 22, + "name": "root_job", + "type_info": "Uuid" + }, + { + "ordinal": 23, "name": "timeout", "type_info": "Int4" }, { - "ordinal": 23, + "ordinal": 24, "name": "flow_step_id", "type_info": "Varchar" }, { - "ordinal": 24, + "ordinal": 25, "name": "cache_ttl", "type_info": "Int4" }, { - "ordinal": 25, + "ordinal": 26, "name": "priority", "type_info": "Int2" }, { - "ordinal": 26, + "ordinal": 27, "name": "preprocessed", "type_info": "Bool" }, { - "ordinal": 27, + "ordinal": 28, "name": "script_entrypoint_override", "type_info": "Varchar" }, { - "ordinal": 28, + "ordinal": 29, "name": "trigger", "type_info": "Varchar" }, { - "ordinal": 29, + "ordinal": 30, "name": "trigger_kind: JobTriggerKind", "type_info": { "Custom": { @@ -233,7 +238,7 @@ } }, { - "ordinal": 30, + "ordinal": 31, "name": "visible_to_owner", "type_info": "Bool" } @@ -274,8 +279,9 @@ true, true, true, + true, false ] }, - "hash": "4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0" + "hash": "f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22" } diff --git a/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json b/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json deleted file mode 100644 index 57ae674eb4..0000000000 --- a/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT flow_version.value->>'early_return' as early_return\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "early_return", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index dfa2a01850..76cf87e58c 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -d16e52d570f10dfdabb04a9061fe7ebdbf5a2aa1 \ No newline at end of file +a5911739330d09cdc084bb277e7e3db8f8d86801 \ No newline at end of file diff --git a/backend/migrations/20250902070305_variable_resource_cache_notifications.down.sql b/backend/migrations/20250902070305_variable_resource_cache_notifications.down.sql new file mode 100644 index 0000000000..7a40964be5 --- /dev/null +++ b/backend/migrations/20250902070305_variable_resource_cache_notifications.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here +DROP TRIGGER IF EXISTS variable_cache_invalidate_trigger ON variable; +DROP TRIGGER IF EXISTS resource_cache_invalidate_trigger ON resource; +DROP FUNCTION IF EXISTS notify_var_resource_cache_change(); \ No newline at end of file diff --git a/backend/migrations/20250902070305_variable_resource_cache_notifications.up.sql b/backend/migrations/20250902070305_variable_resource_cache_notifications.up.sql new file mode 100644 index 0000000000..b96bcc2e50 --- /dev/null +++ b/backend/migrations/20250902070305_variable_resource_cache_notifications.up.sql @@ -0,0 +1,36 @@ +-- Add up migration script here + +-- Create notification function for variable and resource cache invalidation +CREATE OR REPLACE FUNCTION notify_var_resource_cache_change() +RETURNS TRIGGER AS $$ +BEGIN + IF TG_TABLE_NAME = 'variable' THEN + PERFORM pg_notify('var_cache_invalidation', + json_build_object( + 'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id), + 'path', COALESCE(NEW.path, OLD.path), + 'operation', TG_OP + )::text + ); + ELSIF TG_TABLE_NAME = 'resource' THEN + PERFORM pg_notify('resource_cache_invalidation', + json_build_object( + 'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id), + 'path', COALESCE(NEW.path, OLD.path), + 'operation', TG_OP + )::text + ); + END IF; + RETURN COALESCE(NEW, OLD); +END; +$$ LANGUAGE plpgsql; + +-- Create triggers for variable table +CREATE TRIGGER variable_cache_invalidate_trigger + AFTER INSERT OR UPDATE OR DELETE ON variable + FOR EACH ROW EXECUTE FUNCTION notify_var_resource_cache_change(); + +-- Create triggers for resource table +CREATE TRIGGER resource_cache_invalidate_trigger + AFTER INSERT OR UPDATE OR DELETE ON resource + FOR EACH ROW EXECUTE FUNCTION notify_var_resource_cache_change(); \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 1dce88b3dc..9d560a1933 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -473,6 +473,9 @@ async fn windmill_main() -> anyhow::Result<()> { } else { // This time we use a pool of connections let db = windmill_common::connect_db(server_mode, indexer_mode, worker_mode).await?; + + // NOTE: Variable/resource cache initialization moved to API server in windmill-api + Connection::Sql(db) }; @@ -933,6 +936,26 @@ Windmill Community Edition {GIT_VERSION} tracing::info!("Token invalidation detected for token: {}...", &token[..token.len().min(8)]); windmill_api::auth::invalidate_token_from_cache(token); }, + "var_cache_invalidation" => { + if let Ok(payload) = serde_json::from_str::(n.payload()) { + if let (Some(workspace_id), Some(path)) = + (payload.get("workspace_id").and_then(|v| v.as_str()), + payload.get("path").and_then(|v| v.as_str())) { + tracing::info!("Variable cache invalidation detected: {}:{}", workspace_id, path); + windmill_api::var_resource_cache::invalidate_variable_cache(&workspace_id, &path); + } + } + }, + "resource_cache_invalidation" => { + if let Ok(payload) = serde_json::from_str::(n.payload()) { + if let (Some(workspace_id), Some(path)) = + (payload.get("workspace_id").and_then(|v| v.as_str()), + payload.get("path").and_then(|v| v.as_str())) { + tracing::info!("Resource cache invalidation detected: {}:{}", workspace_id, path); + windmill_api::var_resource_cache::invalidate_resource_cache(&workspace_id, &path); + } + } + }, "notify_global_setting_change" => { tracing::info!("Global setting change detected: {}", n.payload()); match n.payload() { @@ -1276,6 +1299,8 @@ async fn listen_pg(url: &str) -> Option { "notify_workspace_key_change", "notify_runnable_version_change", "notify_token_invalidation", + "var_cache_invalidation", + "resource_cache_invalidation", ]; #[cfg(feature = "http_trigger")] diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 85d7555e1b..dcf20ae402 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -949,6 +949,7 @@ impl RunJob { /* schedule_path */ None, /* parent_job */ None, /* root job */ None, + /* flow_innermost_root_job */ None, /* job_id */ None, /* is_flow_step */ false, /* same_worker */ false, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 487e9bae57..9ffa2f9d86 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -14137,6 +14137,10 @@ components: $ref: "#/components/schemas/AIProviderModel" code_completion_model: $ref: "#/components/schemas/AIProviderModel" + custom_prompts: + type: object + additionalProperties: + type: string Alert: type: object diff --git a/backend/windmill-api/src/ai.rs b/backend/windmill-api/src/ai.rs index a41840404d..fc60c65c3c 100644 --- a/backend/windmill-api/src/ai.rs +++ b/backend/windmill-api/src/ai.rs @@ -353,6 +353,8 @@ pub struct AIConfig { pub default_model: Option, #[serde(skip_serializing_if = "Option::is_none")] pub code_completion_model: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub custom_prompts: Option>, } pub fn global_service() -> Router { diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 37743eae34..d20c4052ed 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -1089,6 +1089,7 @@ async fn create_app_internal<'a>( None, None, None, + None, false, false, None, @@ -1467,6 +1468,7 @@ async fn update_app_internal<'a>( None, None, None, + None, false, false, None, @@ -1780,6 +1782,7 @@ async fn execute_component( None, None, None, + None, job_id, false, false, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 9ea840a2d1..4e9b3e4a18 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -510,6 +510,7 @@ async fn create_flow( None, None, None, + None, false, false, None, @@ -977,6 +978,7 @@ async fn update_flow( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5332bcb6c4..b29d683a71 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -332,21 +332,16 @@ async fn get_root_job( Ok(Json(res)) } -async fn compute_root_job_for_flow(db: &DB, w_id: &str, mut job_id: Uuid) -> error::Result { - // TODO: use `root_job` ? - loop { - job_id = match sqlx::query_scalar!( - "SELECT parent_job FROM v2_job WHERE id = $1 AND workspace_id = $2", - job_id, - w_id - ) - .fetch_one(db) - .await - { - Ok(Some(job_id)) => job_id, - _ => return Ok(job_id.to_string()), - } - } +async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result { + let root_job = sqlx::query_scalar!( + r#"SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as "root_job!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#, + job_id, + w_id + ) + .fetch_one(db) + .await?; + + Ok(root_job.to_string()) } async fn get_db_clock(Extension(db): Extension) -> windmill_common::error::JsonResult { @@ -3875,6 +3870,7 @@ pub async fn run_flow_by_path_inner( scheduled_for, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -3970,6 +3966,7 @@ pub async fn restart_flow( scheduled_for, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -4066,6 +4063,7 @@ pub async fn run_script_by_path_inner( scheduled_for, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -4215,6 +4213,7 @@ pub async fn run_workflow_as_code( scheduled_for, None, Some(job_id), + None, job.root_job.or(Some(job_id)), run_query.job_id, false, @@ -4746,6 +4745,7 @@ pub async fn run_wait_result_job_by_path_get( None, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -4897,6 +4897,7 @@ pub async fn run_wait_result_script_by_path_internal( None, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -5012,6 +5013,7 @@ pub async fn run_wait_result_script_by_hash( None, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -5130,6 +5132,7 @@ pub async fn run_wait_result_flow_by_path_internal( scheduled_for, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, @@ -5201,6 +5204,7 @@ async fn run_preview_script( None, None, None, + None, run_query.job_id, false, false, @@ -5312,6 +5316,7 @@ async fn run_bundle_preview_script( None, None, None, + None, run_query.job_id, false, false, @@ -5479,6 +5484,7 @@ async fn run_dependencies_job( None, None, None, + None, false, false, None, @@ -5545,6 +5551,7 @@ async fn run_flow_dependencies_job( None, None, None, + None, false, false, None, @@ -5885,6 +5892,7 @@ async fn run_preview_flow_job( None, None, None, + None, run_query.job_id, false, false, @@ -6024,6 +6032,7 @@ pub async fn run_job_by_hash_inner( scheduled_for, None, run_query.parent_job, + None, run_query.root_job, run_query.job_id, false, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 5b4a04302c..0a083a5161 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -191,6 +191,7 @@ mod users; pub mod users_ee; mod users_oss; mod utils; +pub mod var_resource_cache; mod variables; pub mod webhook_util; #[cfg(feature = "websocket")] @@ -308,6 +309,7 @@ pub async fn run_server( )); let argon2 = Arc::new(Argon2::default()); + let disable_response_logs = std::env::var("DISABLE_RESPONSE_LOGS") .ok() .map(|x| x == "true") diff --git a/backend/windmill-api/src/mcp/server.rs b/backend/windmill-api/src/mcp/server.rs index 8bfe35a009..e51d008e3c 100644 --- a/backend/windmill-api/src/mcp/server.rs +++ b/backend/windmill-api/src/mcp/server.rs @@ -190,9 +190,10 @@ impl ServerHandler for Runner { } // Continue with script/flow logic - let (tool_type, path, is_hub) = - reverse_transform(&request.name).unwrap_or_default(); - + let (tool_type, path, is_hub) = reverse_transform(&request.name).map_err(|e| { + Error::internal_error(format!("Failed to reverse transform path: {}", e), None) + })?; + let item_schema = if is_hub { get_hub_script_schema(&format!("hub/{}", path), db).await? } else { diff --git a/backend/windmill-api/src/mcp/utils/transform.rs b/backend/windmill-api/src/mcp/utils/transform.rs index 6574173bc4..c40fe37263 100644 --- a/backend/windmill-api/src/mcp/utils/transform.rs +++ b/backend/windmill-api/src/mcp/utils/transform.rs @@ -12,16 +12,9 @@ use super::models::SchemaType; /// have names with slashes. Because we replace slashes with underscores, /// we also need to escape underscores. pub fn transform_path(path: &str, type_str: &str) -> String { - // Only apply special underscore escaping for paths starting with "f/" - let transformed = if path.starts_with("f/") { - let escaped_path = path.replace('_', "__"); - escaped_path.replace('/', "_") - } else { - path.replace('/', "_") - }; - + let escaped_path = path.replace('_', "__").replace('/', "_"); // first letter of type_str is used as prefix, only one letter to avoid reaching 60 char name limit - format!("{}-{}", &type_str[..1], transformed) + format!("{}-{}", &type_str[..1], escaped_path) } /// Reverse the transformation of a path @@ -53,19 +46,15 @@ pub fn reverse_transform(transformed_path: &str) -> Result<(&str, String, bool), let mangled_path = &transformed_path[2..]; - // Check if this path was previously transformed with special underscore handling - let is_special_path = mangled_path.starts_with("f_"); - let original_path = if is_hub { let parts = mangled_path.split("-").collect::>(); + if parts.is_empty() { + return Err(format!("Invalid transformed path: {}", transformed_path)); + } parts[0].to_string() - } else if is_special_path { - const TEMP_PLACEHOLDER: &str = "@@UNDERSCORE@@"; - let path_with_placeholder = mangled_path.replace("__", TEMP_PLACEHOLDER); - let path_with_slashes = path_with_placeholder.replace('_', "/"); - path_with_slashes.replace(TEMP_PLACEHOLDER, "_") } else { - mangled_path.replacen('_', "/", 2) + const TEMP_PLACEHOLDER: &str = "@@UNDERSCORE@@"; + mangled_path.replace("__", TEMP_PLACEHOLDER).replace('_', "/").replace(TEMP_PLACEHOLDER, "_") }; Ok((type_str, original_path, is_hub)) diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index b474e710db..f07c3b6760 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -36,6 +36,7 @@ use windmill_common::{ variables, worker::CLOUD_HOSTED, }; +use crate::var_resource_cache::{get_cached_resource, cache_resource}; pub fn workspaced_service() -> Router { Router::new() @@ -326,14 +327,29 @@ async fn exists_resource( Ok(Json(exists)) } +#[derive(Deserialize)] +struct GetResourceQuery { + allow_cache: Option, +} + async fn get_resource_value( authed: ApiAuthed, Extension(user_db): Extension, Extension(db): Extension, + Query(q): Query, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult> { let path = path.to_path(); check_scopes(&authed, || format!("resources:read:{}", path))?; + + // Check cache first when explicitly allowed + let allow_cache = q.allow_cache.unwrap_or(false); + if allow_cache { + if let Some(cached_value) = get_cached_resource(&w_id, &path) { + return Ok(Json(Some(cached_value))); + } + } + let mut tx = user_db.begin(&authed).await?; let value_o = sqlx::query_scalar!( @@ -350,6 +366,14 @@ async fn get_resource_value( } let value = not_found_if_none(value_o, "Resource", path)?; + + // Cache the result if it exists and caching is allowed + if allow_cache { + if let Some(ref val) = value { + cache_resource(&w_id, &path, val.clone()); + } + } + Ok(Json(value)) } @@ -549,12 +573,18 @@ pub async fn transform_json_value<'c>( let job_id = job_id.unwrap(); let job = sqlx::query!( "SELECT - email AS \"email!\", - created_by AS \"created_by!\", - parent_job, permissioned_as AS \"permissioned_as!\", - script_path, schedule_path, flow_step_id, root_job, - scheduled_for AS \"scheduled_for!: chrono::DateTime\" - FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", + v2_job.permissioned_as_email, + v2_job.created_by, + v2_job.parent_job, + v2_job.permissioned_as, + v2_job.runnable_path, + CASE WHEN v2_job.trigger_kind = 'schedule'::job_trigger_kind THEN v2_job.trigger END AS schedule_path, + v2_job.flow_step_id, + v2_job.flow_innermost_root_job, + v2_job.root_job, + v2_job_queue.scheduled_for AS \"scheduled_for: chrono::DateTime\" + FROM v2_job INNER JOIN v2_job_queue ON v2_job.id = v2_job_queue.id + WHERE v2_job.id = $1 AND v2_job.workspace_id = $2", job_id, workspace ) @@ -581,15 +611,16 @@ pub async fn transform_json_value<'c>( &db.into(), workspace, token, - &job.email, + &job.permissioned_as_email, &job.created_by, &job_id.to_string(), &job.permissioned_as, - job.script_path.clone(), + job.runnable_path.clone(), job.parent_job.map(|x| x.to_string()), flow_path, job.schedule_path.clone(), job.flow_step_id.clone(), + job.flow_innermost_root_job.map(|x| x.to_string()), job.root_job.map(|x| x.to_string()), Some(job.scheduled_for.clone()), None, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index f0b2c4c483..9f55d22d65 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -1007,6 +1007,7 @@ async fn create_script_internal<'c>( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/trigger_helpers.rs b/backend/windmill-api/src/trigger_helpers.rs index 1c2bdc029c..e8211d471c 100644 --- a/backend/windmill-api/src/trigger_helpers.rs +++ b/backend/windmill-api/src/trigger_helpers.rs @@ -774,6 +774,7 @@ async fn trigger_script_with_retry_and_error_handler( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/var_resource_cache.rs b/backend/windmill-api/src/var_resource_cache.rs new file mode 100644 index 0000000000..bd35647b87 --- /dev/null +++ b/backend/windmill-api/src/var_resource_cache.rs @@ -0,0 +1,122 @@ +/* + * Author: Claude + * Copyright: Windmill Labs, Inc 2025 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use quick_cache::sync::Cache; +use serde_json::Value; +use std::time::{SystemTime, UNIX_EPOCH}; +use windmill_common::variables::ListableVariable; + +/// Cache TTL for variables and resources (60 seconds) +const CACHE_TTL_SECS: u64 = 60; + +/// Cache entry with timestamp and value (following raw script cache pattern) +#[derive(Clone, Debug)] +pub struct CacheEntry { + pub timestamp: u64, + pub value: T, +} + +impl CacheEntry { + pub fn new(value: T) -> Self { + Self { + timestamp: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(), + value, + } + } + + pub fn is_expired(&self) -> bool { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + now > self.timestamp + CACHE_TTL_SECS + } +} + +lazy_static::lazy_static! { + /// Cache for individual variable values: key = "workspace_id:path" + pub static ref VARIABLE_CACHE: Cache> = Cache::new(1000); + + /// Cache for resource values: key = "workspace_id:path" + pub static ref RESOURCE_CACHE: Cache> = Cache::new(1000); +} + + +/// Generate cache key for variables and resources +pub fn cache_key(workspace_id: &str, path: &str) -> String { + format!("{}:{}", workspace_id, path) +} + +/// Get cached variable if available and not expired +pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option { + let key = cache_key(workspace_id, path); + VARIABLE_CACHE.get(&key).and_then(|entry| { + if entry.is_expired() { + VARIABLE_CACHE.remove(&key); + None + } else { + tracing::debug!("Cache hit for variable {}", key); + Some(entry.value.clone()) + } + }) +} + +/// Cache variable data +pub fn cache_variable(workspace_id: &str, path: &str, variable: ListableVariable) { + let key = cache_key(workspace_id, path); + let entry = CacheEntry::new(variable); + VARIABLE_CACHE.insert(key.clone(), entry); + tracing::debug!("Cached variable {}", key); +} + +/// Get cached resource if available and not expired +pub fn get_cached_resource(workspace_id: &str, path: &str) -> Option { + let key = cache_key(workspace_id, path); + RESOURCE_CACHE.get(&key).and_then(|entry| { + if entry.is_expired() { + RESOURCE_CACHE.remove(&key); + None + } else { + tracing::debug!("Cache hit for resource {}", key); + Some(entry.value.clone()) + } + }) +} + +/// Cache resource data +pub fn cache_resource(workspace_id: &str, path: &str, resource: Value) { + let key = cache_key(workspace_id, path); + let entry = CacheEntry::new(resource); + RESOURCE_CACHE.insert(key.clone(), entry); + tracing::debug!("Cached resource {}", key); +} + +/// Invalidate specific variable from cache +pub fn invalidate_variable_cache(workspace_id: &str, path: &str) { + let key = cache_key(workspace_id, path); + VARIABLE_CACHE.remove(&key); + tracing::info!("Variable cache invalidated for {}", key); +} + +/// Invalidate specific resource from cache +pub fn invalidate_resource_cache(workspace_id: &str, path: &str) { + let key = cache_key(workspace_id, path); + RESOURCE_CACHE.remove(&key); + tracing::info!("Resource cache invalidated for {}", key); +} + +/// Clear all caches (for testing/debugging) +#[allow(dead_code)] +pub fn clear_all_caches() { + VARIABLE_CACHE.clear(); + RESOURCE_CACHE.clear(); + tracing::debug!("All variable/resource caches cleared"); +} \ No newline at end of file diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index f7476d9f26..2250bdfd34 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -39,6 +39,7 @@ use serde::Deserialize; use sqlx::{Postgres, Transaction}; use windmill_common::variables::{decrypt, encrypt}; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; +use crate::var_resource_cache::{get_cached_variable, cache_variable}; lazy_static! { pub static ref SECRET_SALT: Option = std::env::var("SECRET_SALT").ok(); @@ -77,6 +78,7 @@ async fn list_contextual_variables( Some("u/user/triggering_flow_path".to_string()), Some("c".to_string()), Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()), + Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()), Some(chrono::offset::Utc::now()), Some(ScriptHash(1234567890)), ) @@ -133,6 +135,7 @@ async fn list_variables( struct GetVariableQuery { decrypt_secret: Option, include_encrypted: Option, + allow_cache: Option, } async fn get_variable( @@ -144,6 +147,18 @@ async fn get_variable( ) -> JsonResult { let path = path.to_path(); check_scopes(&authed, || format!("variables:read:{}", path))?; + + // Check cache first when explicitly allowed (and for appropriate requests) + let decrypt_secret = q.decrypt_secret.unwrap_or(true); + let allow_cache = q.allow_cache.unwrap_or(false); + let include_encrypted = q.include_encrypted.unwrap_or(false); + + if allow_cache && (!decrypt_secret || include_encrypted) { + if let Some(cached_variable) = get_cached_variable(&w_id, &path) { + return Ok(Json(cached_variable)); + } + } + let mut tx = user_db.begin(&authed).await?; let variable_o = sqlx::query_as::<_, ListableVariable>( @@ -217,6 +232,11 @@ async fn get_variable( variable }; + // Cache the result when explicitly allowed and caching appropriate + if allow_cache && (!decrypt_secret || include_encrypted) { + cache_variable(&w_id, &path, r.clone()); + } + Ok(Json(r)) } diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index fb0ea74f26..c2b9e448c2 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -720,6 +720,7 @@ async fn edit_deploy_to() -> Result { } pub const BANNED_DOMAINS: &str = include_str!("../banned_domains.txt"); +pub const MAX_CUSTOM_PROMPT_LENGTH: usize = 5000; async fn is_allowed_auto_domain(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult { let domain = email.split('@').last().unwrap(); @@ -819,6 +820,20 @@ async fn edit_copilot_config( ) -> Result { require_admin(is_admin, &username)?; + // Validate custom prompts length + if let Some(ref custom_prompts) = ai_config.custom_prompts { + for (mode, prompt) in custom_prompts.iter() { + if prompt.len() > MAX_CUSTOM_PROMPT_LENGTH { + return Err(Error::BadRequest(format!( + "Custom prompt for mode '{}' exceeds maximum length of {} characters (current: {})", + mode, + MAX_CUSTOM_PROMPT_LENGTH, + prompt.len() + ))); + } + } + } + let mut tx = db.begin().await?; sqlx::query!( @@ -887,6 +902,7 @@ async fn get_copilot_info( providers: None, default_model: None, code_completion_model: None, + custom_prompts: None, })) } } diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index f43292755e..ae81512a0a 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -29,7 +29,7 @@ pub struct ContextualVariable { pub is_custom: bool, } -#[derive(Serialize, Deserialize, sqlx::FromRow)] +#[derive(Serialize, Deserialize, sqlx::FromRow, Clone)] pub struct ListableVariable { pub workspace_id: String, @@ -210,7 +210,8 @@ pub async fn get_reserved_variables( flow_path: Option, schedule_path: Option, step_id: Option, - root_flow_id: Option, + flow_innermost_root_job: Option, + root_job_id: Option, scheduled_for: Option>, runnable_id: Option, ) -> Vec { @@ -322,8 +323,14 @@ pub async fn get_reserved_variables( }, ContextualVariable { name: "WM_ROOT_FLOW_JOB_ID".to_string(), - value: root_flow_id.unwrap_or_else(|| "".to_string()), - description: "Job id of the root flow if the job is a flow step".to_string(), + value: flow_innermost_root_job.unwrap_or_else(|| "".to_string()), + description: "Job id of the innermost root flow if the job is a flow step".to_string(), + is_custom: false, + }, + ContextualVariable { + name: "WM_ROOT_JOB_ID".to_string(), + value: root_job_id.unwrap_or_else(|| "".to_string()), + description: "Job id of the root job".to_string(), is_custom: false, }, ContextualVariable { diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7ec4cd09f5..745102bf66 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -380,7 +380,7 @@ fn format_pull_query(peek: String) -> String { id, workspace_id, parent_job, created_by, created_at, runnable_id, runnable_path, args, kind, trigger, trigger_kind, permissioned_as, permissioned_as_email, script_lang, - flow_innermost_root_job, flow_step_id, + flow_innermost_root_job, root_job, flow_step_id, same_worker, pre_run_error, visible_to_owner, tag, concurrent_limit, concurrency_time_window_s, timeout, cache_ttl, priority, raw_code, raw_lock, raw_flow, script_entrypoint_override, preprocessed @@ -391,7 +391,7 @@ fn format_pull_query(peek: String) -> String { canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as, flow_status, j.script_lang, j.same_worker, j.pre_run_error, j.visible_to_owner, - j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, + j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job, j.timeout, j.flow_step_id, j.cache_ttl, j.priority, j.raw_code, j.raw_lock, j.raw_flow, j.script_entrypoint_override, j.preprocessed, pj.runnable_path as parent_runnable_path, COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 3667239119..d84e5cdc97 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -439,6 +439,7 @@ pub async fn push_init_job<'c>( None, None, None, + None, false, true, None, @@ -492,6 +493,7 @@ pub async fn push_periodic_bash_job<'c>( None, None, None, + None, false, true, None, @@ -1267,6 +1269,7 @@ async fn restart_job_if_perpetual_inner( None, None, None, + None, false, false, None, @@ -2029,6 +2032,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( None, None, Some(job_id), + None, Some(job_id), None, false, @@ -2138,6 +2142,7 @@ async fn handle_recovered_schedule<'a, 'c, T: Serialize + Send + Sync>( None, None, Some(job_id), + None, Some(job_id), None, false, @@ -2228,6 +2233,7 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>( None, None, Some(job_id), + None, Some(job_id), None, false, @@ -2311,6 +2317,7 @@ pub struct MiniPulledJob { pub concurrent_limit: Option, pub concurrency_time_window_s: Option, pub flow_innermost_root_job: Option, + pub root_job: Option, pub timeout: Option, pub flow_step_id: Option, pub cache_ttl: Option, @@ -2369,7 +2376,8 @@ impl MiniPulledJob { pre_run_error: job.pre_run_error.clone(), concurrent_limit: job.concurrent_limit.clone(), concurrency_time_window_s: job.concurrency_time_window_s.clone(), - flow_innermost_root_job: job.root_job.clone(), + flow_innermost_root_job: job.root_job.clone(), // QueuedJob is taken from v2_as_queue, where root_job corresponds to flow_innermost_root_job in v2_job + root_job: None, timeout: job.timeout.clone(), flow_step_id: job.flow_step_id.clone(), cache_ttl: job.cache_ttl.clone(), @@ -2574,6 +2582,7 @@ pub async fn get_mini_pulled_job<'c>( concurrent_limit, concurrency_time_window_s, flow_innermost_root_job, + root_job, timeout, flow_step_id, cache_ttl, @@ -3623,6 +3632,7 @@ pub async fn push<'c, 'd>( schedule_path: Option, parent_job: Option, root_job: Option, + flow_innermost_root_job: Option, job_id: Option, _is_flow_step: bool, mut same_worker: bool, // whether the job will be executed on the same worker: if true, the job will be set to running but started_at will not be set. @@ -4683,6 +4693,16 @@ pub async fn push<'c, 'd>( None }; + let root_job = if root_job.is_some() + && (root_job == flow_innermost_root_job.or(parent_job).or(Some(job_id))) + { + // We only save the root job if it's not the innermost root job, parent job, or the job itself as an optimization + // Reference: see [`windmill_worker::common::get_root_job_id`] for logic on determining the root job. + None + } else { + root_job + }; + sqlx::query!( "WITH inserted_job AS ( INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job, @@ -4724,7 +4744,7 @@ pub async fn push<'c, 'd>( pre_run_error.map(|e| e.to_string()), email, visible_to_owner, - root_job, + flow_innermost_root_job, concurrent_limit, if concurrent_limit.is_some() { concurrency_time_window_s @@ -4746,7 +4766,7 @@ pub async fn push<'c, 'd>( job_authed.is_operator, folders.as_slice(), job_authed.groups.as_slice(), - root_job.or(parent_job), + root_job, trigger_kind as Option, running, ) @@ -5137,6 +5157,7 @@ pub async fn get_same_worker_job( v2_job.concurrent_limit, v2_job.concurrency_time_window_s, v2_job.flow_innermost_root_job, + v2_job.root_job, v2_job.timeout, v2_job.flow_step_id, v2_job.cache_ttl, diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index ae3653881a..cd682629c2 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -294,6 +294,7 @@ pub async fn push_scheduled_job<'c>( None, None, None, + None, false, false, None, diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index c8ee996bc7..b9a7cfcd88 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -484,6 +484,7 @@ async fn call_tool( agent_job.schedule_path(), Some(agent_job.id), None, + None, Some(job_id), false, false, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index d0ad991bf2..d64926b580 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1590,6 +1590,7 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 1df44c34df..ce1619da8b 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -211,25 +211,6 @@ pub fn parse_npm_config(s: &str) -> (String, Option) { return (url, token_opt); } -#[async_recursion] -pub async fn get_root_job_id(job: &Uuid, db: &Pool) -> anyhow::Result { - let njob = sqlx::query_scalar!( - "SELECT flow_innermost_root_job FROM v2_job WHERE id = $1", - job - ) - .fetch_optional(db) - .await? - .flatten(); - if let Some(root_job) = njob { - if root_job == *job { - return Ok(job.to_owned()); - } - get_root_job_id(&root_job, db).await - } else { - Ok(job.to_owned()) - } -} - #[async_recursion] pub async fn transform_json_value( name: &str, @@ -272,9 +253,7 @@ pub async fn transform_json_value( Connection::Sql(db) => { let encrypted = y.strip_prefix("$encrypted:").unwrap(); - let root_job_id = - get_root_job_id(&job.flow_innermost_root_job.unwrap_or_else(|| job.id), db) - .await?; + let root_job_id = get_root_job_id(&job); let mc = build_crypt_with_key_suffix( &db, &job.workspace_id, @@ -463,6 +442,7 @@ pub async fn get_reserved_variables( job.schedule_path(), job.flow_step_id.clone(), job.flow_innermost_root_job.clone().map(|x| x.to_string()), + Some(get_root_job_id(job).to_string()), Some(job.scheduled_for.clone()), job.runnable_id, ) @@ -1073,6 +1053,14 @@ pub fn build_http_client(timeout_duration: std::time::Duration) -> error::Result .map_err(|e| Error::internal_err(format!("Error building http client: {e:#}"))) } +pub fn get_root_job_id(job: &MiniPulledJob) -> uuid::Uuid { + // fallback to flow_innermost_root_job and parent_job as root_job is not set if equal to innermost root job or parent job + job.root_job + .or(job.flow_innermost_root_job) + .or(job.parent_job) + .unwrap_or(job.id) +} + #[derive(Clone)] pub struct S3ModeWorkerData { pub client: AuthedClient, diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 607e6375d2..a61b581262 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -545,6 +545,7 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 70f980052b..e8291f771a 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -2146,6 +2146,7 @@ pub async fn start_worker( None, None, None, + None, ) .await .to_vec(); @@ -2266,6 +2267,7 @@ for line in sys.stdin: None, None, None, + None, ) .await; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index f4874f6456..36f255a836 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -11,7 +11,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; -use crate::common::{cached_result_path, save_in_cache}; +use crate::common::{cached_result_path, get_root_job_id, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; use crate::worker_utils::get_tag_and_concurrency; use crate::{ @@ -2816,7 +2816,7 @@ async fn push_next_flow_job( let value_with_parallel = module.get_value_with_parallel()?; - let root_job = if { + let flow_innermost_root_job = if { value_with_parallel.type_ == "flow" || (value_with_parallel.type_ == "forloopflow" && value_with_parallel.parallel.is_some_and(|x| x)) @@ -2828,19 +2828,13 @@ async fn push_next_flow_job( .or_else(|| Some(flow_job.id)) }; + let flow_root_job = get_root_job_id(&flow_job); + // forward root job permissions to the new job - let job_perms: Option = { - if let Some(root_job) = &flow_job - .flow_innermost_root_job - .or_else(|| Some(flow_job.id)) - { - get_job_perms(&mut *tx, root_job, &flow_job.workspace_id) - .await? - .map(|x| x.into()) - } else { - None - } - }; + let job_perms: Option = + get_job_perms(&mut *tx, &flow_root_job, &flow_job.workspace_id) + .await? + .map(|x| x.into()); tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}"); let tag = if !matches!(step, Step::PreprocessorStep) @@ -2860,6 +2854,7 @@ async fn push_next_flow_job( flow_job.permissioned_as.to_owned(), ) }; + let tx2 = PushIsolationLevel::Transaction(tx); let (uuid, mut inner_tx) = push( &db, @@ -2877,7 +2872,8 @@ async fn push_next_flow_job( scheduled_for_o, flow_job.schedule_path(), Some(flow_job.id), - root_job, + Some(flow_root_job), + flow_innermost_root_job, None, true, continue_on_same_worker, @@ -2938,7 +2934,7 @@ async fn push_next_flow_job( SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1) WHERE id = $2", uuid_singleton_json, - root_job.unwrap_or(flow_job.id) + flow_innermost_root_job.unwrap_or(flow_job.id) ) .execute(&mut *inner_tx) .warn_after_seconds(3) diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 358e93e0a1..38a451d8fa 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -739,6 +739,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( None, None, None, + None, false, false, None, diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index dcd91e1d1f..ed7ea56157 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -37,7 +37,8 @@ import { getStringError } from './utils' import type { FlowModuleState, FlowState } from '$lib/components/flows/flowState' import type { CurrentEditor, ExtendedOpenFlow } from '$lib/components/flows/types' import { untrack } from 'svelte' -import { getCurrentModel, type DBSchemas } from '$lib/stores' +import { get } from 'svelte/store' +import { getCurrentModel, type DBSchemas, copilotInfo } from '$lib/stores' import { askTools, prepareAskSystemMessage, prepareAskUserMessage } from './ask/core' import { chatState, DEFAULT_SIZE, triggerablesByAi } from './sharedChatState.svelte' import type { ContextElement } from './context' @@ -208,7 +209,8 @@ class AIChatManager { this.mode = mode this.pendingPrompt = pendingPrompt ?? '' if (mode === AIMode.SCRIPT) { - this.systemMessage = prepareScriptSystemMessage() + const customPrompt = get(copilotInfo).customPrompts?.[mode] + this.systemMessage = prepareScriptSystemMessage(customPrompt) this.systemMessage.content = this.NAVIGATION_SYSTEM_PROMPT + this.systemMessage.content const context = this.contextManager.getSelectedContext() const lang = this.scriptEditorOptions?.lang ?? 'bun' @@ -243,20 +245,24 @@ class AIChatManager { } } } else if (mode === AIMode.FLOW) { - this.systemMessage = prepareFlowSystemMessage() + const customPrompt = get(copilotInfo).customPrompts?.[mode] + this.systemMessage = prepareFlowSystemMessage(customPrompt) this.systemMessage.content = this.NAVIGATION_SYSTEM_PROMPT + this.systemMessage.content this.tools = [this.changeModeTool, ...flowTools] this.helpers = this.flowAiChatHelpers } else if (mode === AIMode.NAVIGATOR) { - this.systemMessage = prepareNavigatorSystemMessage() + const customPrompt = get(copilotInfo).customPrompts?.[mode] + this.systemMessage = prepareNavigatorSystemMessage(customPrompt) this.tools = [this.changeModeTool, ...navigatorTools] this.helpers = {} } else if (mode === AIMode.ASK) { - this.systemMessage = prepareAskSystemMessage() + const customPrompt = get(copilotInfo).customPrompts?.[mode] + this.systemMessage = prepareAskSystemMessage(customPrompt) this.tools = [...askTools] this.helpers = {} } else if (mode === AIMode.API) { - this.systemMessage = prepareApiSystemMessage() + const customPrompt = get(copilotInfo).customPrompts?.[mode] + this.systemMessage = prepareApiSystemMessage(customPrompt) this.tools = [...this.apiTools] this.helpers = {} } diff --git a/frontend/src/lib/components/copilot/chat/api/core.ts b/frontend/src/lib/components/copilot/chat/api/core.ts index cd7038595a..4e47baea72 100644 --- a/frontend/src/lib/components/copilot/chat/api/core.ts +++ b/frontend/src/lib/components/copilot/chat/api/core.ts @@ -57,10 +57,17 @@ export async function getApiTools(): Promise[]> { export const apiTools: Tool<{}>[] = [getDocumentationTool] -export function prepareApiSystemMessage(): ChatCompletionSystemMessageParam { +export function prepareApiSystemMessage(customPrompt?: string): ChatCompletionSystemMessageParam { + let content = CHAT_SYSTEM_PROMPT(get(userStore)?.username ?? '') + + // If there's a custom prompt, append it to the system prompt + if (customPrompt?.trim()) { + content = `${content}\n\nUSER GIVEN INSTRUCTIONS:\n${customPrompt.trim()}` + } + return { role: 'system', - content: CHAT_SYSTEM_PROMPT(get(userStore)?.username ?? '') + content } } diff --git a/frontend/src/lib/components/copilot/chat/ask/core.ts b/frontend/src/lib/components/copilot/chat/ask/core.ts index 997594cc88..f9ba219599 100644 --- a/frontend/src/lib/components/copilot/chat/ask/core.ts +++ b/frontend/src/lib/components/copilot/chat/ask/core.ts @@ -25,10 +25,17 @@ GENERAL PRINCIPLES: export const askTools: Tool<{}>[] = [getDocumentationTool] -export function prepareAskSystemMessage(): ChatCompletionSystemMessageParam { +export function prepareAskSystemMessage(customPrompt?: string): ChatCompletionSystemMessageParam { + let content = CHAT_SYSTEM_PROMPT + + // If there's a custom prompt, append it to the system prompt + if (customPrompt?.trim()) { + content = `${content}\n\nUSER GIVEN INSTRUCTIONS:\n${customPrompt.trim()}` + } + return { role: 'system', - content: CHAT_SYSTEM_PROMPT + content } } diff --git a/frontend/src/lib/components/copilot/chat/flow/core.ts b/frontend/src/lib/components/copilot/chat/flow/core.ts index dae637f156..1fdd8895ff 100644 --- a/frontend/src/lib/components/copilot/chat/flow/core.ts +++ b/frontend/src/lib/components/copilot/chat/flow/core.ts @@ -885,8 +885,8 @@ export const flowTools: Tool[] = [ } ] -export function prepareFlowSystemMessage(): ChatCompletionSystemMessageParam { - const content = `You are a helpful assistant that creates and edits workflows on the Windmill platform. You're provided with a bunch of tools to help you edit the flow. +export function prepareFlowSystemMessage(customPrompt?: string): ChatCompletionSystemMessageParam { + let content = `You are a helpful assistant that creates and edits workflows on the Windmill platform. You're provided with a bunch of tools to help you edit the flow. Follow the user instructions carefully. Go step by step, and explain what you're doing as you're doing it. DO NOT wait for user confirmation before performing an action. Only do it if the user explicitly asks you to wait in their initial instructions. @@ -1012,6 +1012,11 @@ If the user needs a resource as flow input, you should set the property type in If the user wants a specific resource as step input, you should set the step value to a static string in the following format: "$res:path/to/resource". ` + // If there's a custom prompt, append it to the system prompt + if (customPrompt?.trim()) { + content = `${content}\n\nUSER GIVEN INSTRUCTIONS:\n${customPrompt.trim()}` + } + return { role: 'system', content diff --git a/frontend/src/lib/components/copilot/chat/navigator/core.ts b/frontend/src/lib/components/copilot/chat/navigator/core.ts index f7488b3fa8..4c9cf38b70 100644 --- a/frontend/src/lib/components/copilot/chat/navigator/core.ts +++ b/frontend/src/lib/components/copilot/chat/navigator/core.ts @@ -350,10 +350,17 @@ export const navigatorTools: Tool<{}>[] = [ getAvailableResourcesTool ] -export function prepareNavigatorSystemMessage(): ChatCompletionSystemMessageParam { +export function prepareNavigatorSystemMessage(customPrompt?: string): ChatCompletionSystemMessageParam { + let content = CHAT_SYSTEM_PROMPT + + // If there's a custom prompt, append it to the system prompt + if (customPrompt?.trim()) { + content = `${content}\n\nUSER GIVEN INSTRUCTIONS:\n${customPrompt.trim()}` + } + return { role: 'system', - content: CHAT_SYSTEM_PROMPT + content } } diff --git a/frontend/src/lib/components/copilot/chat/script/core.ts b/frontend/src/lib/components/copilot/chat/script/core.ts index 9deb4f0412..ca9bde59e4 100644 --- a/frontend/src/lib/components/copilot/chat/script/core.ts +++ b/frontend/src/lib/components/copilot/chat/script/core.ts @@ -452,10 +452,19 @@ WINDMILL LANGUAGE CONTEXT: ` -export function prepareScriptSystemMessage(): ChatCompletionSystemMessageParam { +export function prepareScriptSystemMessage( + customPrompt?: string +): ChatCompletionSystemMessageParam { + let content = CHAT_SYSTEM_PROMPT + + // If there's a custom prompt, prepend it to the system prompt + if (customPrompt?.trim()) { + content = `${content}\n\nUSER GIVEN INSTRUCTIONS:\n${customPrompt.trim()}` + } + return { role: 'system', - content: CHAT_SYSTEM_PROMPT + content } } diff --git a/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte b/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte index 819765403d..59be11236b 100644 --- a/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte +++ b/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte @@ -4,7 +4,6 @@ import Popover from '$lib/components/Popover.svelte' import { classNames, type StateStore } from '$lib/utils' import { - AlertTriangle, Bed, Database, Gauge, @@ -16,7 +15,8 @@ Pin, X, Play, - Loader2 + Loader2, + TriangleAlert } from 'lucide-svelte' import { createEventDispatcher, getContext } from 'svelte' import { fade } from 'svelte/transition' @@ -460,60 +460,67 @@ {#if deletable && !action} - - - {#if id !== 'preprocessor'} + {#if id !== 'preprocessor'} + + {/if} - {/if} + {#if (id && Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}).length > 0) || Boolean(warningMessage)} -
- - {#snippet text()} -
    - {#if id} - {#each Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}) as m} -
  • - {m.message} -
    • - {/each} - {/if} -
- {/snippet} -
x.type === 'error' - ) - ? 'border-red-600 text-red-600 bg-red-100 hover:bg-red-300' - : 'border-yellow-600 text-yellow-600 bg-yellow-100 hover:bg-yellow-300' - )} - > - -
- -
+ x.type === 'error' + ) + ? 'border-red-600 text-red-600 bg-red-100 hover:bg-red-300' + : ' text-yellow-600 bg-yellow-100 hover:bg-yellow-300' + )} + > + {#snippet text()} +
    + {#if id} + {#each Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}) as m} +
  • + {m.message} +
    • + {/each} + {/if} +
+ {/snippet} + + + {/if} {/if} @@ -521,7 +528,7 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration {#if editMode && enableTestRun && flowJob?.type !== 'QueuedJob'}
(hover = true)} onmouseleave={() => (hover = false)} > @@ -529,11 +536,11 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration
{#if !testIsLoading} {/if} {:else if id?.startsWith('subflow:')} diff --git a/frontend/src/lib/components/graph/renderers/nodes/BranchAllStart.svelte b/frontend/src/lib/components/graph/renderers/nodes/BranchAllStart.svelte index 923a82b13c..09ba0d81d8 100644 --- a/frontend/src/lib/components/graph/renderers/nodes/BranchAllStart.svelte +++ b/frontend/src/lib/components/graph/renderers/nodes/BranchAllStart.svelte @@ -40,9 +40,8 @@ {#if data.insertable}
{/if} + {#if Object.keys(aiProviders).length > 0} +
+

Custom system prompts

+
+ + + +
+
+ {/if} +