starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-25 16:02:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into di/s3-proxy

Browse Source
This commit is contained in:
Diego Imbert
2025-09-02 12:09:52 +02:00
parent ac784a7cd3 fbd942f179
commit dedb43c94f
49 changed files with 709 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json
Generated
+77
View File
@@ -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<chrono::Utc>\"\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<chrono::Utc>",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false,
false,
true,
false,
true,
null,
true,
true,
true,
false
]
},
"hash": "02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0"
}
backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json
Generated
-71
View File
@@ -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<chrono::Utc>\"\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<chrono::Utc>",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true,
true,
true,
true,
true,
true,
true
]
},
"hash": "b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2"
}
backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json → backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json
Generated
+16 -10
View File
@@ -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<HashMap<String, Box<RawValue>>>\",\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<Box<RawValue>>\",\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<HashMap<String, Box<RawValue>>>\",\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<Box<RawValue>>\",\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"
}
backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json
Generated
-23
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
d16e52d570f10dfdabb04a9061fe7ebdbf5a2aa1
a5911739330d09cdc084bb277e7e3db8f8d86801
backend/migrations/20250902070305_variable_resource_cache_notifications.down.sql
+4
View File
@@ -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();
backend/migrations/20250902070305_variable_resource_cache_notifications.up.sql
+36
View File
@@ -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();
backend/src/main.rs
+25
View File
@@ -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::<serde_json::Value>(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::<serde_json::Value>(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<PgListener> {
"notify_workspace_key_change",
"notify_runnable_version_change",
"notify_token_invalidation",
"var_cache_invalidation",
"resource_cache_invalidation",
];
#[cfg(feature = "http_trigger")]
backend/tests/worker.rs
+1
View File
@@ -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,
backend/windmill-api/openapi.yaml
+4
View File
@@ -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
backend/windmill-api/src/ai.rs
+2
View File
@@ -353,6 +353,8 @@ pub struct AIConfig {
pub default_model: Option<ProviderModel>,
#[serde(skip_serializing_if = "Option::is_none")]
pub code_completion_model: Option<ProviderModel>,
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_prompts: Option<HashMap<String, String>>,
}
pub fn global_service() -> Router {
backend/windmill-api/src/apps.rs
+3
View File
@@ -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,
backend/windmill-api/src/flows.rs
+2
View File
@@ -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,
backend/windmill-api/src/jobs.rs
+24 -15
View File
@@ -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<String> {
// 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<String> {
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<DB>) -> windmill_common::error::JsonResult<i64> {
@@ -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,
backend/windmill-api/src/lib.rs
+2
View File
@@ -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")
backend/windmill-api/src/mcp/server.rs
+4 -3
View File
@@ -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 {
backend/windmill-api/src/mcp/utils/transform.rs
+7 -18
View File
@@ -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::<Vec<&str>>();
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))
backend/windmill-api/src/resources.rs
+39 -8
View File
@@ -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<bool>,
}
async fn get_resource_value(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Query(q): Query<GetResourceQuery>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Option<serde_json::Value>> {
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<chrono::Utc>\"
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<chrono::Utc>\"
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,
backend/windmill-api/src/scripts.rs
+1
View File
@@ -1007,6 +1007,7 @@ async fn create_script_internal<'c>(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/trigger_helpers.rs
+1
View File
@@ -774,6 +774,7 @@ async fn trigger_script_with_retry_and_error_handler(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/var_resource_cache.rs
+122
View File
@@ -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<T> {
pub timestamp: u64,
pub value: T,
}
impl<T> CacheEntry<T> {
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<String, CacheEntry<ListableVariable>> = Cache::new(1000);
/// Cache for resource values: key = "workspace_id:path"
pub static ref RESOURCE_CACHE: Cache<String, CacheEntry<Value>> = 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<ListableVariable> {
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<Value> {
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");
}
backend/windmill-api/src/variables.rs
+20
View File
@@ -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<String> = 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<bool>,
include_encrypted: Option<bool>,
allow_cache: Option<bool>,
}
async fn get_variable(
@@ -144,6 +147,18 @@ async fn get_variable(
) -> JsonResult<ListableVariable> {
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))
}
backend/windmill-api/src/workspaces.rs
+16
View File
@@ -720,6 +720,7 @@ async fn edit_deploy_to() -> Result<String> {
}
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<bool> {
let domain = email.split('@').last().unwrap();
@@ -819,6 +820,20 @@ async fn edit_copilot_config(
) -> Result<String> {
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,
}))
}
}
backend/windmill-common/src/variables.rs
+11 -4
View File
@@ -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<String>,
schedule_path: Option<String>,
step_id: Option<String>,
root_flow_id: Option<String>,
flow_innermost_root_job: Option<String>,
root_job_id: Option<String>,
scheduled_for: Option<chrono::DateTime<Utc>>,
runnable_id: Option<ScriptHash>,
) -> Vec<ContextualVariable> {
@@ -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 {
backend/windmill-common/src/worker.rs
+2 -2
View File
@@ -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,
backend/windmill-queue/src/jobs.rs
+24 -3
View File
@@ -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<i32>,
pub concurrency_time_window_s: Option<i32>,
pub flow_innermost_root_job: Option<Uuid>,
pub root_job: Option<Uuid>,
pub timeout: Option<i32>,
pub flow_step_id: Option<String>,
pub cache_ttl: Option<i32>,
@@ -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<String>,
parent_job: Option<Uuid>,
root_job: Option<Uuid>,
flow_innermost_root_job: Option<Uuid>,
job_id: Option<Uuid>,
_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<JobTriggerKind>,
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,
backend/windmill-queue/src/schedule.rs
+1
View File
@@ -294,6 +294,7 @@ pub async fn push_scheduled_job<'c>(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-worker/src/ai_executor.rs
+1
View File
@@ -484,6 +484,7 @@ async fn call_tool(
agent_job.schedule_path(),
Some(agent_job.id),
None,
None,
Some(job_id),
false,
false,
backend/windmill-worker/src/bun_executor.rs
+1
View File
@@ -1590,6 +1590,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/common.rs
+10 -22
View File
@@ -211,25 +211,6 @@ pub fn parse_npm_config(s: &str) -> (String, Option<String>) {
return (url, token_opt);
}
#[async_recursion]
pub async fn get_root_job_id(job: &Uuid, db: &Pool<Postgres>) -> anyhow::Result<Uuid> {
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,
backend/windmill-worker/src/deno_executor.rs
+1
View File
@@ -545,6 +545,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/python_executor.rs
+2
View File
@@ -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;
backend/windmill-worker/src/worker_flow.rs
+12 -16
View File
@@ -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<Authed> = {
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<Authed> =
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)
backend/windmill-worker/src/worker_lockfiles.rs
+1
View File
@@ -739,6 +739,7 @@ pub async fn trigger_dependents_to_recompute_dependencies(
None,
None,
None,
None,
false,
false,
None,
frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts
+12 -6
View File
@@ -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 = {}
}
frontend/src/lib/components/copilot/chat/api/core.ts
+9 -2
View File
@@ -57,10 +57,17 @@ export async function getApiTools(): Promise<Tool<{}>[]> {
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
}
}
frontend/src/lib/components/copilot/chat/ask/core.ts
+9 -2
View File
@@ -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
}
}
frontend/src/lib/components/copilot/chat/flow/core.ts
+7 -2
View File
@@ -885,8 +885,8 @@ export const flowTools: Tool<FlowAIChatHelpers>[] = [
}
]
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
frontend/src/lib/components/copilot/chat/navigator/core.ts
+9 -2
View File
@@ -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
}
}
frontend/src/lib/components/copilot/chat/script/core.ts
+11 -2
View File
@@ -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
}
}
frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte
+62 -55
View File
@@ -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 @@
</div>
{#if deletable && !action}
<button
class="absolute -top-[10px] -right-[10px] rounded-full h-[20px] w-[20px] trash center-center text-secondary
outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration-0 hover:bg-red-400 hover:text-white
{hover || selected ? '' : '!hidden'}"
title="Delete"
onclick={stopPropagation(
preventDefault((event) => dispatch('delete', { id, type: modType }))
)}
onpointerdown={stopPropagation(preventDefault(() => {}))}
<div
class="absolute -translate-y-[100%] top-2 -right-2 flex flex-row gap-1 p-1 min-w-[52px] h-7 group justify-end"
>
<X class="mx-[3px]" size={12} strokeWidth={2} />
</button>
{#if id !== 'preprocessor'}
{#if id !== 'preprocessor'}
<button
class={twMerge(
'trash center-center p-1 text-secondary shadow-sm bg-surface duration-0 hover:bg-blue-400 hover:text-white',
hover ? 'block' : '!hidden',
'shadow-md rounded-md',
'group-hover:block'
)}
onclick={stopPropagation(preventDefault((event) => dispatch('move')))}
title="Move"
>
<Move size={12} />
</button>
{/if}
<button
class="absolute -top-[10px] right-[60px] rounded-full h-[20px] w-[20px] trash center-center text-secondary
outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration-0 hover:bg-blue-400 hover:text-white
{hover ? '' : '!hidden'}"
onclick={stopPropagation(preventDefault((event) => dispatch('move')))}
title="Move"
class={twMerge(
'trash center-center text-secondary shadow-sm bg-surface duration-0 hover:bg-red-400 hover:text-white p-1',
selected || hover ? 'block' : '!hidden',
'group-hover:block',
'shadow-md rounded-md'
)}
title="Delete"
onclick={stopPropagation(
preventDefault((event) => dispatch('delete', { id, type: modType }))
)}
onpointerdown={stopPropagation(preventDefault(() => {}))}
>
<Move class="mx-[3px]" size={12} strokeWidth={2} />
<X size={12} />
</button>
{/if}
</div>
{#if (id && Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}).length > 0) || Boolean(warningMessage)}
<div class="absolute -top-[10px] -left-[10px]">
<Popover>
{#snippet text()}
<ul class="list-disc px-2">
{#if id}
{#each Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}) as m}
<li>
{m.message}
</li>
{/each}
{/if}
</ul>
{/snippet}
<div
class={twMerge(
'flex items-center justify-center h-full w-full rounded-md p-0.5 border duration-0 ',
id &&
Object.values($flowInputsStore?.[id]?.flowStepWarnings || {})?.some(
(x) => 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'
)}
>
<AlertTriangle size={14} strokeWidth={2} />
</div>
</Popover>
</div>
<Popover
class={twMerge(
'absolute -translate-y-[100%] top-1 -left-1',
'flex items-center justify-center rounded-b-none rounded-md p-1 shadow-md duration-0 ',
id &&
Object.values($flowInputsStore?.[id]?.flowStepWarnings || {})?.some(
(x) => 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()}
<ul class="list-disc px-2">
{#if id}
{#each Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}) as m}
<li>
{m.message}
</li>
{/each}
{/if}
</ul>
{/snippet}
<TriangleAlert size={12} strokeWidth={2} />
</Popover>
{/if}
{/if}
</div>
@@ -521,7 +528,7 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration
{#if editMode && enableTestRun && flowJob?.type !== 'QueuedJob'}
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class="absolute top-1/2 -translate-y-1/2 -translate-x-[100%] -left-[0] flex items-center w-fit px-2 h-9 min-w-14"
class="absolute top-1/2 -translate-y-1/2 -translate-x-[100%] -left-[0] flex items-center w-fit px-1 h-9 min-w-9"
onmouseenter={() => (hover = true)}
onmouseleave={() => (hover = false)}
>
@@ -529,11 +536,11 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration
<div transition:fade={{ duration: 100 }}>
{#if !testIsLoading}
<Button
size="sm"
size="xs"
color="light"
title="Run"
variant="border"
btnClasses="p-1.5"
btnClasses="px-1 py-1.5"
on:click={() => {
outputPicker?.toggleOpen(true)
moduleTest?.loadArgsAndRunTest()
@@ -548,7 +555,7 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration
}
}
]}
dropdownBtnClasses="!w-4 px-1"
dropdownBtnClasses="!w-3 px-0.5"
bind:dropdownOpen={testRunDropdownOpen}
>
{#if testIsLoading}
@@ -562,7 +569,7 @@ outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration
size="xs"
color="red"
variant="contained"
btnClasses="!h-[25.5px] !w-[44.5px] !p-1.5 gap-0.5"
btnClasses="!h-[25.5px] !w-[36px] !p-1.5 gap-0.5"
on:click={async () => {
moduleTest?.cancelJob()
}}
frontend/src/lib/components/flows/map/FlowModuleSchemaItemViewer.svelte
+1 -1
View File
@@ -81,7 +81,7 @@ hover:border-blue-700 hover:!visible {hover ? '' : '!hidden'}"
onclick?.()
})
)}
title="Edit Id"><Pencil size={14} /></button
title="Edit Id"><Pencil size={12} /></button
>
{/if}
{:else if id?.startsWith('subflow:')}
frontend/src/lib/components/graph/renderers/nodes/BranchAllStart.svelte
+2 -3
View File
@@ -40,9 +40,8 @@
{#if data.insertable}
<button
title="Delete branch"
class="z-50 absolute -top-[10px] -right-[10px] rounded-full h-[20px] w-[20px] center-center text-primary
outline-[1px] outline outline-gray-700 bg-surface duration-0 hover:bg-red-400 hover:text-white
hover:border-red-700"
class="z-50 absolute -translate-y-[100%] top-1 -right-1 rounded-md p-1 center-center text-primary
bg-surface duration-0 hover:bg-red-400 hover:text-white shadow-md"
onclick={stopPropagation(
preventDefault(() => {
data.eventHandlers.deleteBranch(
frontend/src/lib/components/graph/renderers/nodes/BranchOneStart.svelte
+2 -3
View File
@@ -36,9 +36,8 @@
{#if data.insertable}
<button
title="Delete branch"
class="z-50 absolute -top-[10px] -right-[10px] rounded-full h-[20px] w-[20px] center-center text-primary
border-[1.5px] border-gray-700 bg-surface duration-0 hover:bg-red-400 hover:text-white
hover:border-red-700"
class="z-50 absolute -translate-y-[100%] top-1 -right-1 rounded-md p-1 center-center text-primary
bg-surface duration-0 hover:bg-red-400 hover:text-white shadow-md"
onclick={stopPropagation(
preventDefault(() => {
data.eventHandlers.deleteBranch(
frontend/src/lib/components/settings/TokenDisplay.svelte
+1 -1
View File
@@ -18,7 +18,7 @@
}: Props = $props()
function handleCopyClick() {
copyToClipboard(token)
copyToClipboard(mcpUrl || token)
onCopy?.()
}
frontend/src/lib/components/workspaceSettings/AISettings.svelte
+70 -1
View File
@@ -14,6 +14,10 @@
import { safeSelectItems } from '../select/utils.svelte'
import Badge from '../common/badge/Badge.svelte'
import Tooltip from '../Tooltip.svelte'
import { AIMode } from '../copilot/chat/AIChatManager.svelte'
import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte'
import autosize from '$lib/autosize'
const aiProviderLabels: [AIProvider, string][] = [
['openai', 'OpenAI'],
@@ -28,15 +32,19 @@
['customai', 'Custom AI']
]
const MAX_CUSTOM_PROMPT_LENGTH = 5000
let {
aiProviders = $bindable(),
codeCompletionModel = $bindable(),
defaultModel = $bindable(),
customPrompts = $bindable(),
usingOpenaiClientCredentialsOauth = $bindable()
}: {
aiProviders: Exclude<AIConfig['providers'], undefined>
codeCompletionModel: string | undefined
defaultModel: string | undefined
customPrompts: Record<string, string>
usingOpenaiClientCredentialsOauth: boolean
} = $props()
@@ -47,6 +55,9 @@
) as Record<AIProvider, string[]>
)
// Custom system prompt settings
let selectedAiMode = $state<AIMode>(AIMode.ASK)
let selectedAiModels = $derived(Object.values(aiProviders).flatMap((p) => p.models))
let modelProviderMap = $derived(
Object.fromEntries(
@@ -94,10 +105,16 @@
defaultModel && modelProviderMap[defaultModel]
? { model: defaultModel, provider: modelProviderMap[defaultModel] }
: undefined
// Convert customPrompts to include only non-empty prompts
const custom_prompts: Record<string, string> = Object.entries(customPrompts)
.filter(([_, prompt]) => prompt.trim().length > 0)
.reduce((acc, [mode, prompt]) => ({ ...acc, [mode]: prompt }), {})
const config: AIConfig = {
providers: aiProviders,
code_completion_model,
default_model
default_model,
custom_prompts: Object.keys(custom_prompts).length > 0 ? custom_prompts : undefined
}
await WorkspaceService.editCopilotConfig({
workspace: $workspaceStore!,
@@ -313,6 +330,58 @@
</div>
{/if}
{#if Object.keys(aiProviders).length > 0}
<div class="flex flex-col gap-2">
<p class="font-semibold">Custom system prompts</p>
<div class="flex flex-col gap-4">
<Label label="AI Mode">
<ToggleButtonGroup
bind:selected={selectedAiMode}
on:selected={({ detail }) => {
selectedAiMode = detail
}}
>
{#snippet children({ item })}
{#each Object.values(AIMode) as mode}
<div class="relative">
<ToggleButton
value={mode}
label={mode.charAt(0).toUpperCase() + mode.slice(1)}
{item}
/>
{#if customPrompts[mode]?.length > 0}
<div
class="absolute -top-1 -right-1 w-2 h-2 bg-blue-500 rounded-full border border-surface"
></div>
{/if}
</div>
{/each}
{/snippet}
</ToggleButtonGroup>
</Label>
<Label
label="Custom system prompt for {selectedAiMode.charAt(0).toUpperCase() +
selectedAiMode.slice(1)} Mode"
>
<textarea
bind:value={customPrompts[selectedAiMode]}
placeholder="Enter a custom system prompt for {selectedAiMode} mode."
class="w-full min-h-24 p-2 border border-gray-200 dark:border-gray-700 rounded-md bg-surface text-primary resize-y"
rows="4"
maxlength={MAX_CUSTOM_PROMPT_LENGTH}
use:autosize
></textarea>
<div class="flex justify-end mt-1">
<span class="text-xs text-secondary">
{(customPrompts[selectedAiMode] ?? '').length}/{MAX_CUSTOM_PROMPT_LENGTH} characters
</span>
</div>
</Label>
</div>
</div>
{/if}
<Button
wrapperClasses="self-start"
disabled={!Object.values(aiProviders).every((p) => p.resource_path) ||
frontend/src/lib/stores.ts
+7 -3
View File
@@ -100,11 +100,13 @@ export const copilotInfo = writable<{
codeCompletionModel?: AIProviderModel
defaultModel?: AIProviderModel
aiModels: AIProviderModel[]
customPrompts?: Record<string, string>
}>({
enabled: false,
codeCompletionModel: undefined,
defaultModel: undefined,
aiModels: []
aiModels: [],
customPrompts: {}
})
export async function loadCopilot(workspace: string) {
@@ -139,7 +141,8 @@ export function setCopilotInfo(aiConfig: AIConfig) {
enabled: true,
codeCompletionModel: aiConfig.code_completion_model,
defaultModel: aiConfig.default_model,
aiModels: aiModels
aiModels: aiModels,
customPrompts: aiConfig.custom_prompts ?? {}
})
} else {
copilotSessionModel.set(undefined)
@@ -148,7 +151,8 @@ export function setCopilotInfo(aiConfig: AIConfig) {
enabled: false,
codeCompletionModel: undefined,
defaultModel: undefined,
aiModels: []
aiModels: [],
customPrompts: {}
})
}
}
frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
+11 -5
View File
@@ -19,9 +19,7 @@
ResourceService,
SettingService,
type AIConfig,
type ErrorHandler
} from '$lib/gen'
import {
enterpriseLicense,
@@ -57,6 +55,7 @@
convertDucklakeSettingsFromBackend,
type DucklakeSettingsType
} from '$lib/components/workspaceSettings/DucklakeSettings.svelte'
import { AIMode } from '$lib/components/copilot/chat/AIChatManager.svelte'
let slackInitialPath: string = $state('')
let slackScriptPath: string = $state('')
@@ -81,6 +80,7 @@
let aiProviders: Exclude<AIConfig['providers'], undefined> = $state({})
let codeCompletionModel: string | undefined = $state(undefined)
let defaultModel: string | undefined = $state(undefined)
let customPrompts: Record<string, string> = $state({})
let s3ResourceSettings: S3ResourceSettings = $state({
resourceType: 's3',
@@ -245,7 +245,12 @@
aiProviders = settings.ai_config?.providers ?? {}
defaultModel = settings.ai_config?.default_model?.model
codeCompletionModel = settings.ai_config?.code_completion_model?.model
customPrompts = settings.ai_config?.custom_prompts ?? {}
for (const mode of Object.values(AIMode)) {
if (!(mode in customPrompts)) {
customPrompts[mode] = ''
}
}
errorHandlerItemKind = settings.error_handler
? (settings.error_handler.split('/')[0] as 'flow' | 'script')
: 'script'
@@ -318,7 +323,7 @@
requestBody: {
error_handler: `${errorHandlerItemKind}/${errorHandlerScriptPath}`,
error_handler_extra_args: errorHandlerExtraArgs,
error_handler_muted_on_cancel: errorHandlerMutedOnCancel,
error_handler_muted_on_cancel: errorHandlerMutedOnCancel
}
})
sendUserToast(`workspace error handler set to ${errorHandlerScriptPath}`)
@@ -328,7 +333,7 @@
requestBody: {
error_handler: undefined,
error_handler_extra_args: undefined,
error_handler_muted_on_cancel: undefined,
error_handler_muted_on_cancel: undefined
}
})
sendUserToast(`workspace error handler removed`)
@@ -804,6 +809,7 @@
bind:aiProviders
bind:codeCompletionModel
bind:defaultModel
bind:customPrompts
bind:usingOpenaiClientCredentialsOauth
/>
{:else if tab == 'windmill_lfs'}