fix sqlx

* feat: workspace dedicated workers

* ref

* chore: update ee-repo-ref to a18ac31062ac092cb9a5fc87629e217d97f4911d

This commit updates the EE repository reference after PR #398 was merged in windmill-ee-private.

Previous ee-repo-ref: 98cfe3fef764d9d815d326d5056c734a03689d33

New ee-repo-ref: a18ac31062ac092cb9a5fc87629e217d97f4911d

Automated by sync-ee-ref workflow.

* fix(frontend): workspace script in flow steps

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

.claude/settings.json

@@ -1,39 +1,4 @@
 {
-  "hooks": {
-    "UserPromptSubmit": [
-      {
-        "hooks": [
-          {
-            "type": "command",
-            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/resolve-symlinks.sh",
-            "timeout": 30
-          }
-        ]
-      }
-    ],
-    "Stop": [
-      {
-        "hooks": [
-          {
-            "type": "command",
-            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/restore-symlinks.sh",
-            "timeout": 30
-          }
-        ]
-      }
-    ],
-    "SessionEnd": [
-      {
-        "hooks": [
-          {
-            "type": "command",
-            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/restore-symlinks.sh",
-            "timeout": 30
-          }
-        ]
-      }
-    ]
-  },
   "permissions": {
     "allow": [
       "Bash(ls:*)",
@@ -93,8 +58,45 @@
     ]
   },
   "enableAllProjectMcpServers": true,
+  "hooks": {
+    "UserPromptSubmit": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/resolve-symlinks.sh",
+            "timeout": 30
+          }
+        ]
+      }
+    ],
+    "Stop": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/restore-symlinks.sh",
+            "timeout": 30
+          }
+        ]
+      }
+    ],
+    "SessionEnd": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/restore-symlinks.sh",
+            "timeout": 30
+          }
+        ]
+      }
+    ]
+  },
   "enabledPlugins": {
     "rust-analyzer-lsp@claude-plugins-official": true,
-    "typescript-lsp@claude-plugins-official": true
+    "typescript-lsp@claude-plugins-official": true,
+    "code-review@claude-plugins-official": true,
+    "commit-commands@claude-plugins-official": true
   }
 }

.claude/skills/rust-backend/SKILL.md

@@ -0,0 +1,495 @@
+---
+name: rust-backend
+description: Rust coding guidelines for the Windmill backend. Apply when writing or modifying Rust code in the backend directory.
+---
+
+# Rust Backend Coding Guidelines
+
+Apply these patterns when writing or modifying Rust code in the `backend/` directory.
+
+## Data Structure Design
+
+Choose between `struct`, `enum`, or `newtype` based on domain needs:
+
+- Use `enum` for state machines instead of boolean flags or loosely related fields
+- Model invariants explicitly using types (e.g., `NonZeroU32`, `Duration`, custom enums)
+- Consider ownership of each field:
+  - Use `&str` vs `String`, slices vs vectors
+  - Use `Arc<T>` when sharing across threads
+  - Use `Cow<'a, T>` for flexible ownership
+
+```rust
+// State machine with enum
+enum JobState {
+    Pending { scheduled_for: DateTime<Utc> },
+    Running { started_at: DateTime<Utc>, worker: String },
+    Completed { result: JobResult, duration_ms: i64 },
+    Failed { error: String, retries: u32 },
+}
+
+// Avoid multiple booleans
+struct Job {
+    is_pending: bool,   // Don't do this
+    is_running: bool,
+    is_completed: bool,
+}
+```
+
+## Impl Block Organization
+
+Place `impl` blocks immediately below the struct/enum they modify. Group methods logically:
+
+```rust
+struct JobQueue {
+    jobs: Vec<Job>,
+    capacity: usize,
+}
+
+impl JobQueue {
+    // Constructors first
+    pub fn new(capacity: usize) -> Self { ... }
+    pub fn with_jobs(jobs: Vec<Job>) -> Self { ... }
+
+    // Getters
+    pub fn len(&self) -> usize { ... }
+    pub fn is_empty(&self) -> bool { ... }
+
+    // Mutation methods
+    pub fn push(&mut self, job: Job) -> Result<()> { ... }
+    pub fn pop(&mut self) -> Option<Job> { ... }
+
+    // Domain logic
+    pub fn next_scheduled(&self) -> Option<&Job> { ... }
+}
+```
+
+## Iterator Chains Over For-Loops
+
+Prefer functional iterator chains (`.filter().map().collect()`) over imperative for-loops:
+
+```rust
+// Preferred
+let results: Vec<_> = items
+    .iter()
+    .filter(|item| item.is_valid())
+    .map(|item| item.transform())
+    .collect();
+
+// Avoid
+let mut results = Vec::new();
+for item in items.iter() {
+    if item.is_valid() {
+        results.push(item.transform());
+    }
+}
+```
+
+## Error Handling
+
+Use the `Error` type from `windmill_common::error`. Return `Result<T, Error>` or `JsonResult<T>` for fallible functions:
+
+```rust
+use windmill_common::error::{Error, Result};
+
+// Use ? operator for propagation
+pub async fn get_job(db: &DB, id: Uuid) -> Result<Job> {
+    let job = sqlx::query_as!(Job, "SELECT ... WHERE id = $1", id)
+        .fetch_optional(db)
+        .await?
+        .ok_or_else(|| Error::NotFound("job not found".to_string()))?;
+    Ok(job)
+}
+```
+
+Prefer `if let` for optional handling. Use `let...else` when early return makes code clearer:
+
+```rust
+let Some(config) = get_config() else {
+    return Err(Error::MissingConfig);
+};
+```
+
+Never panic in library code. Reserve `.unwrap()` for cases with compile-time guarantees. Keep functions short to help lifetime inference and clarity.
+
+## Early Returns
+
+Return early to avoid deep nesting. Handle error cases and edge conditions first:
+
+```rust
+// Preferred - early returns
+fn process_job(job: Option<Job>) -> Result<Output> {
+    let Some(job) = job else {
+        return Ok(Output::default());
+    };
+
+    if !job.is_valid() {
+        return Err(Error::InvalidJob);
+    }
+
+    if job.is_cached() {
+        return Ok(job.cached_result());
+    }
+
+    // Main logic at the end, not nested
+    execute_job(job)
+}
+
+// Avoid - deep nesting
+fn process_job(job: Option<Job>) -> Result<Output> {
+    if let Some(job) = job {
+        if job.is_valid() {
+            if !job.is_cached() {
+                execute_job(job)
+            } else {
+                Ok(job.cached_result())
+            }
+        } else {
+            Err(Error::InvalidJob)
+        }
+    } else {
+        Ok(Output::default())
+    }
+}
+```
+
+## Variable Shadowing
+
+Shadow variables instead of creating new names with prefixes:
+
+```rust
+// Preferred
+let data = fetch_raw_data();
+let data = parse(data);
+let data = validate(data)?;
+
+// Avoid
+let raw_data = fetch_raw_data();
+let parsed_data = parse(raw_data);
+let validated_data = validate(parsed_data)?;
+```
+
+## Minimal Comments
+
+- No inline comments explaining obvious code
+- No TODO/FIXME comments in committed code
+- Doc comments (`///`) only on public items
+- Let code be self-documenting through clear naming
+
+## Type Safety
+
+Use enums over boolean flags for clarity:
+
+```rust
+// Preferred
+enum JobStatus {
+    Pending,
+    Running,
+    Completed,
+}
+
+// Avoid
+struct Job {
+    is_running: bool,
+    is_completed: bool,
+}
+```
+
+## Pattern Matching
+
+Prefer explicit matching. Use wildcards strategically for fallback cases or ignored fields:
+
+```rust
+// Explicit matching preferred
+match status {
+    JobStatus::Pending => handle_pending(),
+    JobStatus::Running => handle_running(),
+    JobStatus::Completed => handle_completed(),
+}
+
+// Wildcards OK for fallback
+match result {
+    Ok(value) => process(value),
+    Err(_) => return default_value(),
+}
+
+// Wildcards OK for ignoring fields in destructuring
+let Point { x, y, .. } = point;
+```
+
+## Destructuring in Function Signatures
+
+Destructure structs directly in function parameters:
+
+```rust
+// Preferred
+async fn process_job(
+    Extension(db): Extension<DB>,
+    Path((workspace, job_id)): Path<(String, Uuid)>,
+    Query(pagination): Query<Pagination>,
+) -> Result<Json<Job>> {
+    // ...
+}
+
+// Avoid
+async fn process_job(
+    db_ext: Extension<DB>,
+    path: Path<(String, Uuid)>,
+    query: Query<Pagination>,
+) -> Result<Json<Job>> {
+    let Extension(db) = db_ext;
+    let Path((workspace, job_id)) = path;
+    // ...
+}
+```
+
+## Trait Implementations
+
+Use standard trait implementations to simplify conversions and reduce boilerplate:
+
+```rust
+// Implement From/Into for type conversions
+impl From<DbJob> for ApiJob {
+    fn from(db: DbJob) -> Self {
+        ApiJob {
+            id: db.id,
+            status: db.status.into(),
+        }
+    }
+}
+
+// Use TryFrom for fallible conversions
+impl TryFrom<String> for JobKind {
+    type Error = Error;
+    fn try_from(s: String) -> Result<Self, Self::Error> { ... }
+}
+```
+
+Apply `derive` macros to reduce boilerplate:
+
+```rust
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Job { ... }
+```
+
+## Module Structure
+
+- Use `pub(crate)` instead of `pub` when possible; expose only what needs exposing
+- Keep APIs small and expressive; avoid leaking internal types
+- Organize code into modules reflecting ownership and domain boundaries
+
+```rust
+// Prefer restricted visibility
+pub(crate) fn internal_helper() { ... }
+
+// Only pub for external API
+pub fn create_job(...) -> Result<Job> { ... }
+```
+
+## Code Navigation
+
+Always use rust-analyzer LSP for:
+- Go to definition
+- Find references
+- Type information
+- Import resolution
+
+Do not guess at module paths or type definitions.
+
+## JSON Handling
+
+Prefer `Box<serde_json::value::RawValue>` over `serde_json::Value` when:
+- Storing JSON in the database (JSONB columns)
+- Passing JSON through without modification
+- The JSON structure doesn't need inspection
+
+```rust
+// Preferred - avoids parsing/serialization overhead
+pub struct Job {
+    pub id: Uuid,
+    pub args: Option<Box<serde_json::value::RawValue>>,
+}
+
+// Only use Value when you need to inspect/modify JSON
+let value: serde_json::Value = serde_json::from_str(&json)?;
+if let Some(field) = value.get("field") {
+    // modify or inspect
+}
+```
+
+## Serde Optimizations
+
+Use serde attributes to optimize serialization:
+
+```rust
+#[derive(Serialize, Deserialize)]
+pub struct Job {
+    #[serde(rename = "jobId")]
+    pub id: Uuid,
+
+    #[serde(default)]
+    pub priority: i32,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub parent_job: Option<Uuid>,
+
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub tags: Vec<String>,
+}
+```
+
+Prefer borrowing for zero-copy deserialization when lifetimes allow:
+
+```rust
+#[derive(Deserialize)]
+pub struct JobInput<'a> {
+    #[serde(borrow)]
+    pub workspace_id: Cow<'a, str>,
+
+    #[serde(borrow)]
+    pub script_path: &'a str,
+}
+```
+
+## SQLx Patterns
+
+**Never use `SELECT *`** - always list columns explicitly. This is critical for backwards compatibility when workers run behind the API server version:
+
+```rust
+// Preferred - explicit columns
+sqlx::query_as!(
+    Job,
+    "SELECT id, workspace_id, path, created_at FROM v2_job WHERE id = $1",
+    job_id
+)
+
+// Avoid - breaks when columns are added
+sqlx::query_as!(Job, "SELECT * FROM v2_job WHERE id = $1", job_id)
+```
+
+Use batch operations to minimize round trips:
+
+```rust
+// Preferred - single query with multiple values
+sqlx::query!(
+    "INSERT INTO job_logs (job_id, logs) VALUES ($1, $2), ($3, $4)",
+    id1, log1, id2, log2
+)
+
+// Avoid N+1 queries
+for id in ids {
+    sqlx::query!("SELECT ... WHERE id = $1", id).fetch_one(db).await?;
+}
+
+// Preferred - single query with IN clause
+sqlx::query!("SELECT ... WHERE id = ANY($1)", &ids[..]).fetch_all(db).await?
+```
+
+Use transactions for multi-step operations and parameterize all queries.
+
+## Async & Tokio Patterns
+
+Never block the async runtime. Use `spawn_blocking` for CPU-intensive or blocking I/O:
+
+```rust
+// Preferred - offload blocking work
+let result = tokio::task::spawn_blocking(move || {
+    expensive_computation(&data)
+}).await?;
+
+// Avoid - blocks the runtime
+let result = expensive_computation(&data);  // Don't do this in async
+```
+
+Use tokio primitives for sleep and channels:
+
+```rust
+use tokio::sync::mpsc;
+use tokio::time::sleep;
+
+// Avoid in async contexts
+use std::thread::sleep; // Blocks the runtime
+```
+
+Use bounded channels for backpressure:
+
+```rust
+// Preferred - bounded channel prevents overwhelming
+let (tx, rx) = tokio::sync::mpsc::channel(100);
+
+// Be careful with unbounded
+let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+```
+
+## Mutex Selection in Async Code
+
+**Prefer `std::sync::Mutex` (or `parking_lot::Mutex`) over `tokio::sync::Mutex`** for protecting data in async code. The async mutex is more expensive and only needed when holding locks across `.await` points.
+
+```rust
+// Preferred for data protection - std mutex is faster
+use std::sync::Mutex;
+
+struct Cache {
+    data: Mutex<HashMap<String, Value>>,
+}
+
+impl Cache {
+    fn get(&self, key: &str) -> Option<Value> {
+        self.data.lock().unwrap().get(key).cloned()
+    }
+
+    fn insert(&self, key: String, value: Value) {
+        self.data.lock().unwrap().insert(key, value);
+    }
+}
+```
+
+**Use `tokio::sync::Mutex` only when you must hold the lock across `.await` points**, typically for IO resources like database connections:
+
+```rust
+use tokio::sync::Mutex;
+use std::sync::Arc;
+
+// Async mutex for IO resources held across await points
+let conn = Arc::new(Mutex::new(db_connection));
+
+async fn execute_query(conn: Arc<Mutex<DbConn>>, query: &str) {
+    let mut lock = conn.lock().await;
+    lock.execute(query).await;  // Lock held across .await
+}
+```
+
+**Common pattern**: Wrap `Arc<Mutex<...>>` in a struct with non-async methods that lock internally, keeping lock scope minimal:
+
+```rust
+struct SharedState {
+    inner: std::sync::Mutex<StateInner>,
+}
+
+impl SharedState {
+    fn update(&self, value: i32) {
+        self.inner.lock().unwrap().value = value;
+    }
+
+    fn get(&self) -> i32 {
+        self.inner.lock().unwrap().value
+    }
+}
+```
+
+**Alternative for IO resources**: Spawn a dedicated task to manage the resource and communicate via message passing:
+
+```rust
+let (tx, mut rx) = tokio::sync::mpsc::channel(32);
+
+tokio::spawn(async move {
+    while let Some(cmd) = rx.recv().await {
+        handle_io_command(&mut resource, cmd).await;
+    }
+});
+```
+
+## Build & Tooling
+
+Build speed tips:
+- Use `cargo check` during rapid iteration over `cargo build`
+- Minimize unnecessary dependencies and feature flags

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## [1.621.2](https://github.com/windmill-labs/windmill/compare/v1.621.1...v1.621.2) (2026-01-29)
+
+
+### Bug Fixes
+
+* **cli:** revert findCodebase change that broke ../shared codebases ([#7740](https://github.com/windmill-labs/windmill/issues/7740)) ([20357f4](https://github.com/windmill-labs/windmill/commit/20357f41f55ce246220ec56ef257ea7d6ac82e3a))
+* do not quit indexer when receiving handoff during pull ([#7659](https://github.com/windmill-labs/windmill/issues/7659)) ([8bb6b63](https://github.com/windmill-labs/windmill/commit/8bb6b6331b74d43b1ecfa08d3393254f54a94f87))
+
 ## [1.621.1](https://github.com/windmill-labs/windmill/compare/v1.621.0...v1.621.1) (2026-01-29)
 
 

backend/.sqlx/query-02bf9763298f301d4fc75490c070a0663142d4d23a2df007361622b94d4783e1.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT flow_version.value AS \"value!: sqlx::types::Json<Box<sqlx::types::JsonRawValue>>\" \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",
+  "query": "SELECT flow_version.value AS \"value!: sqlx::types::Json<Box<sqlx::types::JsonRawValue>>\"\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": [
       {
@@ -19,5 +19,5 @@
       false
     ]
   },
-  "hash": "bbce3e1eae78c48409d4204cd6cb3b9db088f6e51bea5e74a494c4e9f4c3b78e"
+  "hash": "02bf9763298f301d4fc75490c070a0663142d4d23a2df007361622b94d4783e1"
 }

backend/.sqlx/query-97c61b6a9a5112ea484565236959a544511d5d501fb737da8110a8725b883465.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT (worker)\n        DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "TextArray",
+        "Varchar",
+        "Varchar",
+        "TextArray",
+        "Varchar",
+        "Int8",
+        "Int8",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "97c61b6a9a5112ea484565236959a544511d5d501fb737da8110a8725b883465"
+}

backend/.sqlx/query-ed90b9fcc57f530bab2d2d7426795bb358c91ebdd9dab50d7e3d2fdce63b947c.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (worker)\n        DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "TextArray",
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Int8",
-        "Int8",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "ed90b9fcc57f530bab2d2d7426795bb358c91ebdd9dab50d7e3d2fdce63b947c"
-}

backend/CLAUDE.md

@@ -1,22 +1,64 @@
 # Backend Development (Rust)
 
-## Core Principles
+## Project Structure
 
-- Follow @rust-best-practices.mdc for detailed guidelines
-- Database schema reference: @summarized_schema.txt
-- The API routes prefixes are all listed in windmill-api/src/lib.rs
-- This repository is the open source side of the project. The enterprise files (\*\_ee.rs) are in the `windmill-ee-private` folder (a sibling directory). Those files are symlinked into their corresponding locations within each crate's `src/` directory.
+Windmill uses a workspace-based architecture with multiple crates:
 
-## JSON Handling
+- **windmill-api**: API server functionality
+- **windmill-worker**: Job execution
+- **windmill-common**: Shared code used by all crates
+- **windmill-queue**: Job & flow queuing
+- **windmill-audit**: Audit logging
+- Other specialized crates (git-sync, autoscaling, etc.)
 
-- **Prefer `Box<serde_json::value::RawValue>` over `serde_json::Value`** when possible, especially:
-  - When storing JSON in the database (JSONB columns)
-  - When passing JSON through without modification
-  - When the JSON structure doesn't need to be inspected or manipulated
-- This avoids unnecessary parsing/serialization overhead and preserves the original JSON format
-- Use `serde_json::Value` only when you need to inspect, modify, or construct JSON programmatically
+## Key References
 
-## Adding New Features
+- Database schema: @summarized_schema.txt
+- API route prefixes: `windmill-api/src/lib.rs`
 
-1. Update database schema with migration if necessary
-2. Update backend/windmill-api/openapi.yaml after modifying API endpoints
+## Adding New Code
+
+### Module Organization
+
+- Place new code in the appropriate crate based on functionality
+- For API endpoints, create or modify files in `windmill-api/src/` organized by domain
+- For shared functionality, use `windmill-common/src/`
+- Follow existing patterns for file structure and organization
+
+### API Endpoints
+
+- Follow existing patterns in the `windmill-api` crate
+- Use axum's routing system and extractors
+- Update `backend/windmill-api/openapi.yaml` after modifying API endpoints
+
+### Database Changes
+
+- Update database schema with migration if necessary
+- Use `sqlx` for database operations with prepared statements
+- Use transactions for multi-step operations
+
+## Enterprise Features
+
+- Enterprise files use the `*_ee.rs` suffix
+- Enterprise source is in `windmill-ee-private` folder (sibling directory), symlinked into each crate's `src/`
+- Use feature flags: `#[cfg(feature = "enterprise")]`
+- Isolate enterprise code in separate modules
+
+## Testing
+
+- Write unit tests for core functionality
+- Use the `#[cfg(test)]` module for test code
+- For database tests, use the existing test utilities
+
+## Common Crates
+
+- **tokio**: Async runtime
+- **axum**: Web server and routing
+- **sqlx**: Database operations
+- **serde**: Serialization/deserialization
+- **tracing**: Logging and diagnostics
+- **reqwest**: HTTP client
+
+## Coding Guidelines
+
+Detailed Rust coding patterns and best practices are provided by the `rust-backend` skill.

backend/Cargo.lock

@@ -2265,9 +2265,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "4.5.55"
+version = "4.5.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e34525d5bbbd55da2bb745d34b36121baac88d07619a9a09cfcf4a6c0832785"
+checksum = "a75ca66430e33a14957acc24c5077b503e7d374151b2b4b3a10c83b4ceb4be0e"
 dependencies = [
  "clap_builder",
  "clap_derive",
@@ -2275,9 +2275,9 @@ dependencies = [
 
 [[package]]
 name = "clap_builder"
-version = "4.5.55"
+version = "4.5.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "59a20016a20a3da95bef50ec7238dbd09baeef4311dcdd38ec15aba69812fb61"
+checksum = "793207c7fa6300a0608d1080b858e5fdbe713cdc1c8db9fb17777d8a13e63df0"
 dependencies = [
  "anstream",
  "anstyle",
@@ -15466,7 +15466,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
 
 [[package]]
 name = "windmill"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "aws-sdk-config",
@@ -15529,7 +15529,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "argon2",
@@ -15659,7 +15659,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api-client"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "reqwest 0.12.28",
  "serde",
@@ -15669,7 +15669,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-audit"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "chrono",
  "lazy_static",
@@ -15683,7 +15683,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-autoscaling"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "axum 0.7.9",
@@ -15702,7 +15702,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-common"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -15798,7 +15798,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-git-sync"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "regex",
  "serde",
@@ -15813,7 +15813,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-indexer"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "astral-tokio-tar",
@@ -15837,7 +15837,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-macros"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "itertools 0.14.0",
  "lazy_static",
@@ -15853,7 +15853,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-mcp"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -15873,7 +15873,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-oauth"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-oauth2",
@@ -15897,7 +15897,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "convert_case 0.6.0",
  "serde",
@@ -15906,7 +15906,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-bash"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -15918,7 +15918,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-csharp"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "serde_json",
@@ -15930,7 +15930,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-go"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "gosyn",
@@ -15942,7 +15942,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-graphql"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -15954,7 +15954,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-java"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "serde_json",
@@ -15966,7 +15966,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-nu"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "nu-parser",
@@ -15977,7 +15977,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-php"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "itertools 0.14.0",
@@ -15988,7 +15988,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "itertools 0.14.0",
@@ -16001,7 +16001,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py-imports"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -16025,7 +16025,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ruby"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16039,7 +16039,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-rust"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "convert_case 0.6.0",
@@ -16056,7 +16056,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-sql"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16070,7 +16070,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ts"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16089,7 +16089,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-yaml"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "serde",
@@ -16100,7 +16100,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-queue"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -16137,7 +16137,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-sql-datatype-parser-wasm"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-test",
@@ -16147,7 +16147,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-worker"
-version = "1.621.1"
+version = "1.621.2"
 dependencies = [
  "anyhow",
  "async-once-cell",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.621.1"
+version = "1.621.2"
 authors.workspace = true
 edition.workspace = true
 
@@ -35,7 +35,7 @@ members = [
 exclude = ["./windmill-duckdb-ffi-internal"]
 
 [workspace.package]
-version = "1.621.1"
+version = "1.621.2"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 

backend/ee-repo-ref.txt

@@ -1 +1 @@
-5d841b358dd32130c9f34b54f59b96b5c322f213
+a18ac31062ac092cb9a5fc87629e217d97f4911d

backend/migrations/20260129161851_dedicated_workers_column.down.sql

@@ -0,0 +1 @@
+ALTER TABLE worker_ping DROP COLUMN IF EXISTS dedicated_workers;

backend/migrations/20260129161851_dedicated_workers_column.up.sql

@@ -0,0 +1 @@
+ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS dedicated_workers TEXT[];

backend/rust-best-practices.mdc

@@ -1,111 +0,0 @@
----
-description: 
-globs: backend/**/*.rs
-alwaysApply: false
----
-# Windmill Backend - Rust Best Practices
-
-## Project Structure
-
-Windmill uses a workspace-based architecture with multiple crates:
-
-- **windmill-api**: API server functionality
-- **windmill-worker**: Job execution
-- **windmill-common**: Shared code used by all crates
-- **windmill-queue**: Job & flow queuing
-- **windmill-audit**: Audit logging
-- Other specialized crates (git-sync, autoscaling, etc.)
-
-## Adding New Code
-
-### Module Organization
-
-- Place new code in the appropriate crate based on functionality
-- For API endpoints, create or modify files in `windmill-api/src/` organized by domain
-- For shared functionality, use `windmill-common/src/`
-- Use the `_ee.rs` suffix for enterprise-only modules
-- Follow existing patterns for file structure and organization
-
-### Error Handling
-
-- Use the custom `Error` enum from `windmill-common::error`
-- Return `Result<T, Error>` or `JsonResult<T>` for functions that can fail
-- Use the `?` operator for error propagation
-- Add location tracking to errors using `#[track_caller]`
-
-### Database Operations
-
-- Use `sqlx` for database operations with prepared statements
-- Leverage existing database helper functions in `db.rs` modules
-- Use transactions for multi-step operations
-- Handle database errors properly
-
-### API Endpoints
-
-- Follow existing patterns in the `windmill-api` crate
-- Use axum's routing system and extractors
-- Group related routes together
-- Use consistent response formats (JSON)
-- Follow proper authentication and authorization patterns
-- Do not forget to update backend/windmill-api/openapi.yaml after modifying an api endpoint
-
-## Performance Optimizations
-
-When generating code, especially involving `serde`, `sqlx`, and `tokio`, prioritize performance by applying the following principles:
-
-### Serde Optimizations (Serialization & Deserialization)
-
--   **Specify Structure Explicitly:** When defining structs for Serde (`#[derive(Serialize, Deserialize)]`), use `#[serde(...` attributes extensively. This includes:
-    *   `#[serde(rename = "...")]` or `#[serde(alias = "...")]` to map external names precisely, avoiding dynamic lookups.
-    *   `#[serde(default)]` for optional fields with default values, reducing parsing complexity.
-    *   `#[serde(skip_serializing_if = "...")]` to avoid writing fields that meet a certain condition (e.g., `Option::is_none()`, `Vec::is_empty()`, or a custom function), reducing output size and serialization work.
-    *   `#[serde(skip_serializing)]` or `#[serde(skip_deserializing)]` for fields that should *not* be included.
--   **Prefer Borrowing:** Where possible and safe (data lifetime allows), use `Cow<'a, str>` or `&'a str` (with `#[serde(borrow)]`) instead of `String` for string fields during deserialization. This avoids allocating new strings, enabling zero-copy reading from the input buffer. Apply this principle to byte slices (`&'a [u8]` / `Cow<'a, [u8]>`) and potentially borrowed vectors as well.
--   **Avoid Intermediate `Value`:** Unless the data structure is truly dynamic or unknown at compile time, deserialize directly into a well-defined struct or enum rather than into `serde_json::Value` (or equivalent for other formats). This avoids unnecessary heap allocations and type switching.
-
-### SQLx Optimizations (Database Interaction)
-
--   **CRITICAL - Never Use `SELECT *` in Worker-Executed Queries:** For any query that can potentially be executed by workers, **always** explicitly list the specific columns you need instead of using `SELECT *`. This is essential for backwards compatibility: when workers are running behind the API server version (common in distributed deployments), adding new columns to database tables will cause outdated workers to fail when they try to deserialize rows with unexpected columns. Always use explicit column lists like `SELECT id, workspace_id, path, created_at FROM table` instead of `SELECT * FROM table`.
--   **Select Only Necessary Columns:** In `SELECT` queries, list specific column names rather than using `SELECT *`. This reduces data transferred from the database and the work needed for hydration/deserialization.
--   **Batch Operations:** For multiple `INSERT`, `UPDATE`, or `DELETE` statements, prefer executing them in a single query if the database and driver support it efficiently (e.g., `INSERT INTO ... VALUES (...), (...), ...`). This minimizes round trips to the database.
--   **Avoid N+1 Queries:** Do not loop through results of one query and execute a separate query for each item (e.g., fetching users, then querying for each user's profile in a loop). Instead, use JOINs or a single query with an `IN` clause to fetch related data efficiently.
--   **Deserialize Directly:** Use `#[derive(FromRow)]` on structs and ensure the struct fields match the selected columns in the query. This allows SQLx to hydrate objects directly, avoiding intermediate data structures.
--   **Parameterize Queries:** Always use SQLx's query methods (`.bind(...)`) to pass values as parameters rather than string formatting. This prevents SQL injection and allows the database to cache query plans, improving performance on repeated executions.
-
-### Tokio Optimizations (Asynchronous Runtime)
-
--   **Avoid Blocking Operations:** **Crucially**, never perform blocking operations (synchronous file I/O, `std::thread::sleep`, CPU-bound loops, `std::sync::Mutex::lock`, blocking network calls without `tokio::net`) directly within an `async fn` or a standard `tokio::spawn` task. Blocking pauses the entire worker thread, potentially starving other tasks. Use `tokio::task::spawn_blocking` for CPU-intensive work or blocking I/O.
--   **Use Tokio's Async Primitives:** Prefer `tokio::sync` (channels, mutexes, semaphores), `tokio::io`, `tokio::net`, and `tokio::time` over their `std` counterparts in asynchronous contexts. These are designed to yield control back to the scheduler.
--   **Manage Concurrency:** Be mindful of how many tasks are spawned. Creating a new task for every tiny piece of work can introduce overhead. Group related asynchronous operations where appropriate.
--   **Handle Shared State Efficiently:** Use `Arc` for shared ownership in concurrent tasks. When shared state needs mutation, prefer `tokio::sync::Mutex` over `std::sync::Mutex` in `async` code. Consider `tokio::sync::RwLock` if reads significantly outnumber writes. Minimize the duration for which locks are held.
--   **Understand `.await`:** Place `.await` strategically to allow the runtime to switch to other ready tasks. Ensure that `.await` points to genuinely asynchronous operations.
--   **Backpressure:** If dealing with data streams or queues between tasks, implement backpressure mechanisms (e.g., bounded channels like `tokio::sync::mpsc::channel`) to prevent one component from overwhelming another or critical resources like the database.
-
-## Enterprise Features
-
--   Use feature flags for enterprise functionality
--   Conditionally compile with `#[cfg(feature = "enterprise")]`
--   Isolate enterprise code in separate modules
-
-## Code Style
-
--   Group imports by external and internal crates
--   Place struct/enum definitions before implementations
--   Group similar functionality together
--   Use descriptive naming consistent with the codebase
--   Follow existing patterns for async code using tokio
-
-## Testing
-
--   Write unit tests for core functionality
--   Use the `#[cfg(test)]` module for test code
--   For database tests, use the existing test utilities
-
-## Common Crates Used
-
--   **tokio**: For async runtime
--   **axum**: For web server and routing
--   **sqlx**: For database operations
--   **serde**: For serialization/deserialization
--   **tracing**: For logging and diagnostics
--   **reqwest**: For HTTP client functionality

backend/src/main.rs

@@ -909,9 +909,11 @@ Windmill Community Edition {GIT_VERSION}
                     tracing::info!("Received killpill, aborting index initialization");
                 },
                 res = windmill_indexer::completed_runs_oss::init_index(&db) => {
-                        let res = res?;
-                        reader = Some(res.0);
-                        writer = Some(res.1);
+                    let res = res?;
+                    if let Some(r) = res {
+                        reader = Some(r.0);
+                        writer = Some(r.1);
+                    }
                 }
 
                 }
@@ -953,9 +955,11 @@ Windmill Community Edition {GIT_VERSION}
                         tracing::info!("Received killpill, aborting index initialization");
                     },
                     res = windmill_indexer::service_logs_oss::init_index(&db, killpill_tx.clone()) => {
-                            let res = res?;
-                            reader = Some(res.0);
-                            writer = Some(res.1);
+                        let res = res?;
+                        if let Some(r) = res {
+                            reader = Some(r.0);
+                            writer = Some(r.1);
+                        }
                     }
 
                 }

backend/src/monitor.rs

@@ -56,8 +56,8 @@ use windmill_common::{
         HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING,
         JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
         LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING,
-        OTEL_TRACING_PROXY_SETTING, NUGET_CONFIG_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING,
-        POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
+        NUGET_CONFIG_SETTING, OTEL_SETTING, OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING,
+        POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
         REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
         SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
     },
@@ -84,10 +84,10 @@ use windmill_common::{
 use windmill_common::{client::AuthedClient, global_settings::APP_WORKSPACED_ROUTE_SETTING};
 use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload};
 use windmill_worker::{
-    handle_job_error, JobCompletedSender, OtelTracingProxySettings, SameWorkerSender, BUNFIG_INSTALL_SCOPES,
-    INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS,
-    OTEL_TRACING_PROXY_SETTINGS, NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL,
-    PIP_INDEX_URL, POWERSHELL_REPO_PAT, POWERSHELL_REPO_URL,
+    handle_job_error, JobCompletedSender, OtelTracingProxySettings, SameWorkerSender,
+    BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS,
+    NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, OTEL_TRACING_PROXY_SETTINGS,
+    PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, POWERSHELL_REPO_PAT, POWERSHELL_REPO_URL,
 };
 
 #[cfg(feature = "parquet")]
@@ -246,6 +246,7 @@ pub async fn initial_load(
                     ),
                     priority_tags_sorted: vec![],
                     dedicated_worker: None,
+                    dedicated_workers: None,
                     init_bash: load_init_bash_from_env(),
                     periodic_script_bash: load_periodic_bash_script_from_env(),
                     periodic_script_interval_seconds: load_periodic_bash_script_interval_from_env(),
@@ -784,26 +785,24 @@ pub async fn load_keep_job_dir(conn: &Connection) {
 
 pub async fn reload_otel_tracing_proxy_setting(conn: &Connection) {
     match load_value_from_global_settings_with_conn(conn, OTEL_TRACING_PROXY_SETTING, true).await {
-        Ok(Some(settings)) => {
-            match serde_json::from_value::<OtelTracingProxySettings>(settings) {
-                Ok(new_settings) => {
-                    let mut current = OTEL_TRACING_PROXY_SETTINGS.write().await;
-                    if current.enabled != new_settings.enabled
-                        || current.enabled_languages != new_settings.enabled_languages
-                    {
-                        tracing::info!(
-                            "OTEL tracing proxy settings changed: enabled={}, languages={:?}",
-                            new_settings.enabled,
-                            new_settings.enabled_languages
-                        );
-                        *current = new_settings;
-                    }
-                }
-                Err(e) => {
-                    tracing::error!("Error parsing OTEL tracing proxy settings: {e:#}");
+        Ok(Some(settings)) => match serde_json::from_value::<OtelTracingProxySettings>(settings) {
+            Ok(new_settings) => {
+                let mut current = OTEL_TRACING_PROXY_SETTINGS.write().await;
+                if current.enabled != new_settings.enabled
+                    || current.enabled_languages != new_settings.enabled_languages
+                {
+                    tracing::info!(
+                        "OTEL tracing proxy settings changed: enabled={}, languages={:?}",
+                        new_settings.enabled,
+                        new_settings.enabled_languages
+                    );
+                    *current = new_settings;
                 }
             }
-        }
+            Err(e) => {
+                tracing::error!("Error parsing OTEL tracing proxy settings: {e:#}");
+            }
+        },
         Err(e) => {
             tracing::error!("Error loading OTEL tracing proxy setting: {e:#}");
         }
@@ -985,7 +984,10 @@ pub async fn delete_expired_items(db: &DB) -> () {
                 );
             }
         }
-        Err(e) => tracing::error!("Error deleting expired MCP OAuth authorization codes: {:?}", e),
+        Err(e) => tracing::error!(
+            "Error deleting expired MCP OAuth authorization codes: {:?}",
+            e
+        ),
     }
 
     let job_retention_secs = *JOB_RETENTION_SECS.read().await;
@@ -1866,7 +1868,13 @@ pub async fn monitor_db(
 
     let update_min_worker_version_f = async {
         #[cfg(not(feature = "test_job_debouncing"))]
-        windmill_common::min_version::update_min_version(conn, _worker_mode, WORKERS_NAMES.read().await.clone(), initial_load).await;
+        windmill_common::min_version::update_min_version(
+            conn,
+            _worker_mode,
+            WORKERS_NAMES.read().await.clone(),
+            initial_load,
+        )
+        .await;
     };
 
     // Run every 5 minutes (10 iterations * 30s = 5 minutes)
@@ -2060,10 +2068,12 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
     } else {
         let wc = WORKER_CONFIG.read().await;
         let config = config.unwrap();
-        if *wc != config || config.dedicated_worker.is_some() {
+        let has_dedicated = config.dedicated_worker.is_some() || config.dedicated_workers.as_ref().is_some_and(|dws| !dws.is_empty());
+        if *wc != config || has_dedicated {
             if kill_if_change {
-                if config.dedicated_worker.is_some()
+                if has_dedicated
                     || (*wc).dedicated_worker != config.dedicated_worker
+                    || (*wc).dedicated_workers != config.dedicated_workers
                 {
                     tracing::info!("Dedicated worker config changed, sending killpill. Expecting to be restarted by supervisor.");
                     let _ = tx.send();

backend/summarized_schema.txt

@@ -995,6 +995,7 @@ TABLE: worker_ping
   - custom_tags (text[])
   - worker_group (character)
   - dedicated_worker (character)
+  - dedicated_workers (text[])
   - wm_version (character)
   - current_job_id (uuid)
   - current_job_workspace_id (character)

backend/windmill-api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 
 info:
-  version: 1.621.1
+  version: 1.621.2
   title: Windmill API
 
   contact:
@@ -5751,6 +5751,14 @@ paths:
             If true, the description field will be omitted from the response.
           schema:
             type: boolean
+        - name: dedicated_worker
+          in: query
+          description: |
+            (default regardless)
+            If true, show only scripts with dedicated_worker enabled.
+            If false, show only scripts with dedicated_worker disabled.
+          schema:
+            type: boolean
 
       responses:
         "200":
@@ -7178,6 +7186,14 @@ paths:
             If true, the description field will be omitted from the response.
           schema:
             type: boolean
+        - name: dedicated_worker
+          in: query
+          description: |
+            (default regardless)
+            If true, show only flows with dedicated_worker enabled.
+            If false, show only flows with dedicated_worker disabled.
+          schema:
+            type: boolean
       responses:
         "200":
           description: All flow

backend/windmill-api/src/ai.rs

@@ -16,7 +16,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{json, value::RawValue};
 use std::collections::HashMap;
 use windmill_audit::{audit_oss::audit_log, ActionKind};
-use windmill_common::ai_providers::{AIProvider, ProviderConfig, ProviderModel};
+use windmill_common::ai_providers::{empty_string_as_none, AIProvider, ProviderConfig, ProviderModel};
 use windmill_common::error::{to_anyhow, Error, Result};
 use windmill_common::utils::configure_client;
 use windmill_common::variables::get_variable_or_self;
@@ -143,15 +143,17 @@ enum AnthropicPlatform {
 
 #[derive(Deserialize, Debug)]
 struct AIStandardResource {
-    #[serde(alias = "baseUrl")]
+    #[serde(alias = "baseUrl", default, deserialize_with = "empty_string_as_none")]
     base_url: Option<String>,
-    #[serde(alias = "apiKey")]
+    #[serde(alias = "apiKey", default, deserialize_with = "empty_string_as_none")]
     api_key: Option<String>,
+    #[serde(default, deserialize_with = "empty_string_as_none")]
     organization_id: Option<String>,
+    #[serde(default, deserialize_with = "empty_string_as_none")]
     region: Option<String>,
-    #[serde(alias = "awsAccessKeyId")]
+    #[serde(alias = "awsAccessKeyId", default, deserialize_with = "empty_string_as_none")]
     aws_access_key_id: Option<String>,
-    #[serde(alias = "awsSecretAccessKey")]
+    #[serde(alias = "awsSecretAccessKey", default, deserialize_with = "empty_string_as_none")]
     aws_secret_access_key: Option<String>,
     /// Platform for Anthropic API (standard or google_vertex_ai)
     #[serde(default)]
@@ -207,9 +209,14 @@ impl AIRequestConfig {
             AIResource::Standard(resource) => {
                 let region = resource.region.clone();
                 let platform = resource.platform.clone();
-                let base_url = provider
-                    .get_base_url(resource.base_url, resource.region, db)
-                    .await?;
+                // Skip get_base_url for Bedrock - it uses SDK directly, not HTTP
+                let base_url = if matches!(provider, AIProvider::AWSBedrock) {
+                    String::new()
+                } else {
+                    provider
+                        .get_base_url(resource.base_url, db)
+                        .await?
+                };
                 let api_key = if let Some(api_key) = resource.api_key {
                     Some(get_variable_or_self(api_key, db, w_id).await?)
                 } else {
@@ -251,7 +258,7 @@ impl AIRequestConfig {
                     None
                 };
                 let token = Self::get_token_using_oauth(resource, db, w_id).await?;
-                let base_url = provider.get_base_url(None, None, db).await?;
+                let base_url = provider.get_base_url(None, db).await?;
 
                 (
                     None,
@@ -578,7 +585,7 @@ async fn global_proxy(
         return Err(Error::BadRequest("API key is required".to_string()));
     };
 
-    let base_url = provider.get_base_url(None, None, &db).await?;
+    let base_url = provider.get_base_url(None, &db).await?;
 
     let url = format!("{}/{}", base_url, ai_path);
 
@@ -745,7 +752,7 @@ async fn proxy(
             let region = request_config
                 .region
                 .as_deref()
-                .ok_or_else(|| Error::internal_err("AWS region must be set for Bedrock"))?;
+                .unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
 
             // Audit log before making the SDK request
             let mut tx = db.begin().await?;

backend/windmill-api/src/flows.rs

@@ -191,6 +191,9 @@ async fn list_flows(
     if !lq.include_draft_only.unwrap_or(false) || authed.is_operator {
         sqlb.and_where("o.draft_only IS NOT TRUE");
     }
+    if let Some(dw) = &lq.dedicated_worker {
+        sqlb.and_where_eq("dedicated_worker", dw);
+    }
 
     if lq.with_deployment_msg.unwrap_or(false) {
         sqlb.join("deployment_metadata dm")

backend/windmill-api/src/scripts.rs

@@ -371,6 +371,9 @@ async fn list_scripts(
     if let Some(it) = &lq.is_template {
         sqlb.and_where_eq("is_template", it);
     }
+    if let Some(dw) = &lq.dedicated_worker {
+        sqlb.and_where_eq("dedicated_worker", dw);
+    }
     if authed.is_operator {
         sqlb.and_where_eq("kind", quote("script"));
     } else if let Some(lowercased_kinds) = lowercased_kinds {

backend/windmill-common/src/ai_providers.rs

@@ -4,7 +4,17 @@
 
 use crate::db::DB;
 use crate::error::{Error, Result};
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize};
+
+/// Deserializes an Option<String> where empty strings become None.
+/// Use with `#[serde(default, deserialize_with = "empty_string_as_none")]`
+pub fn empty_string_as_none<'de, D>(deserializer: D) -> std::result::Result<Option<String>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let opt = Option::<String>::deserialize(deserializer)?;
+    Ok(opt.filter(|s| !s.is_empty()))
+}
 
 lazy_static::lazy_static! {
     static ref OPENAI_AZURE_BASE_PATH: Option<String> = std::env::var("OPENAI_AZURE_BASE_PATH").ok();
@@ -13,6 +23,10 @@ lazy_static::lazy_static! {
 pub const OPENAI_BASE_URL: &str = "https://api.openai.com/v1";
 pub const GOOGLE_AI_BASE_URL: &str = "https://generativelanguage.googleapis.com/v1beta";
 
+/// Empty string signals BedrockClient::from_env() to use the region from AWS environment/config
+/// (e.g., AWS_REGION or AWS_DEFAULT_REGION env vars, or ~/.aws/config)
+pub const USE_ENV_REGION: &str = "";
+
 #[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Hash, Clone)]
 #[serde(rename_all = "lowercase")]
 pub enum AIProvider {
@@ -36,11 +50,9 @@ impl AIProvider {
     pub async fn get_base_url(
         &self,
         resource_base_url: Option<String>,
-        region: Option<String>,
         db: &DB,
     ) -> Result<String> {
-        // If a base URL is provided in the resource, use it (ignore empty strings)
-        if let Some(base_url) = resource_base_url.filter(|s| !s.is_empty()) {
+        if let Some(base_url) = resource_base_url {
             return Ok(base_url);
         }
 
@@ -78,23 +90,10 @@ impl AIProvider {
                 format!("{:?} provider requires a base URL in the resource", p),
             )),
             AIProvider::AWSBedrock => {
-                #[cfg(feature = "bedrock")]
-                {
-                    Ok(format!(
-                        "https://bedrock-runtime.{}.amazonaws.com",
-                        region
-                            .filter(|s| !s.is_empty())
-                            .unwrap_or_else(|| "us-east-1".to_string())
-                    ))
-                }
-                #[cfg(not(feature = "bedrock"))]
-                {
-                    let _ = region;
-                    Err(Error::BadRequest(
-                        "AWS Bedrock support is not enabled. Build with 'bedrock' feature."
-                            .to_string(),
-                    ))
-                }
+                // AWS Bedrock uses the SDK directly, not HTTP base URL
+                Err(Error::internal_err(
+                    "AWS Bedrock uses SDK directly, not HTTP base URL".to_string(),
+                ))
             }
         }
     }

backend/windmill-common/src/flows.rs

@@ -1146,6 +1146,7 @@ pub struct ListFlowQuery {
     pub starred_only: Option<bool>,
     pub include_draft_only: Option<bool>,
     pub with_deployment_msg: Option<bool>,
+    pub dedicated_worker: Option<bool>,
 }
 
 pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {

backend/windmill-common/src/scripts.rs

@@ -687,6 +687,7 @@ pub struct ListScriptQuery {
     pub with_deployment_msg: Option<bool>,
     #[serde(default, deserialize_with = "from_seq")]
     pub languages: Option<Vec<ScriptLang>>,
+    pub dedicated_worker: Option<bool>,
 }
 
 fn from_seq<'de, D>(deserializer: D) -> Result<Option<Vec<ScriptLang>>, D::Error>

backend/windmill-common/src/worker.rs

@@ -219,6 +219,7 @@ lazy_static::lazy_static! {
         worker_tags: Default::default(),
         priority_tags_sorted: Default::default(),
         dedicated_worker: Default::default(),
+        dedicated_workers: Default::default(),
         cache_clear: Default::default(),
         init_bash: Default::default(),
         periodic_script_bash: Default::default(),
@@ -1243,6 +1244,7 @@ pub struct Ping {
     pub ip: Option<String>,
     pub tags: Option<Vec<String>>,
     pub dw: Option<String>,
+    pub dws: Option<Vec<String>>,
     pub version: Option<String>,
     pub vcpus: Option<i64>,
     pub memory: Option<i64>,
@@ -1298,6 +1300,7 @@ pub async fn update_ping_http(
                 &insert_ping.ip.unwrap(),
                 insert_ping.tags.unwrap_or_default().as_slice(),
                 insert_ping.dw,
+                insert_ping.dws.as_deref(),
                 &insert_ping.version.unwrap(),
                 insert_ping.vcpus,
                 insert_ping.memory,
@@ -1428,6 +1431,7 @@ pub async fn insert_ping_query(
     ip: &str,
     tags: &[String],
     dw: Option<String>,
+    dws: Option<&[String]>,
     version: &str,
     vcpus: Option<i64>,
     memory: Option<i64>,
@@ -1435,14 +1439,15 @@ pub async fn insert_ping_query(
     db: &DB,
 ) -> anyhow::Result<()> {
     sqlx::query!(
-        "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (worker)
-        DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group",
+        "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT (worker)
+        DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers",
         worker_instance,
         worker_name,
         ip,
         tags,
         worker_group,
         dw,
+        dws,
         version,
         vcpus,
         memory,
@@ -1614,6 +1619,30 @@ pub async fn load_worker_config(
             }
         })
         .transpose()?;
+
+    // Parse dedicated_workers (multiple dedicated workers)
+    let dedicated_workers = config
+        .dedicated_workers
+        .map(|workers| {
+            workers
+                .into_iter()
+                .map(|x| {
+                    let splitted = x.split(':').to_owned().collect_vec();
+                    if splitted.len() != 2 {
+                        return Err(anyhow::anyhow!(
+                            "Invalid dedicated_workers format. Got {x}, expects <workspace_id>:<path>"
+                        ));
+                    }
+                    let workspace = splitted[0];
+                    let script_path = splitted[1];
+                    Ok(WorkspacedPath {
+                        workspace_id: workspace.to_string(),
+                        path: script_path.to_string(),
+                    })
+                })
+                .collect::<Result<Vec<_>, _>>()
+        })
+        .transpose()?;
     if *WORKER_GROUP == "default" && dedicated_worker.is_none() {
         let mut all_tags = config
             .worker_tags
@@ -1647,7 +1676,18 @@ pub async fn load_worker_config(
     let worker_tags = config
         .worker_tags
         .or_else(|| {
-            if let Some(ref dedicated_worker) = dedicated_worker.as_ref() {
+            // Check for multiple dedicated workers first
+            if let Some(ref dws) = dedicated_workers.as_ref() {
+                let mut dedi_tags: Vec<String> = dws
+                    .iter()
+                    .map(|dw| format!("{}:{}", dw.workspace_id, dw.path))
+                    .collect();
+                if std::env::var("ADD_FLOW_TAG").is_ok() {
+                    dedi_tags.push("flow".to_string());
+                }
+                Some(dedi_tags)
+            } else if let Some(ref dedicated_worker) = dedicated_worker.as_ref() {
+                // Fallback to single dedicated worker for backward compatibility
                 let mut dedi_tags = vec![format!(
                     "{}:{}",
                     dedicated_worker.workspace_id, dedicated_worker.path
@@ -1743,6 +1783,7 @@ pub async fn load_worker_config(
         worker_tags,
         priority_tags_sorted,
         dedicated_worker,
+        dedicated_workers,
         init_bash: config
             .init_bash
             .or_else(|| load_init_bash_from_env())
@@ -1836,6 +1877,7 @@ pub struct WorkerConfigOpt {
     pub worker_tags: Option<Vec<String>>,
     pub priority_tags: Option<HashMap<String, u8>>,
     pub dedicated_worker: Option<String>,
+    pub dedicated_workers: Option<Vec<String>>,
     pub init_bash: Option<String>,
     pub periodic_script_bash: Option<String>,
     pub periodic_script_interval_seconds: Option<u64>,
@@ -1852,6 +1894,7 @@ impl Default for WorkerConfigOpt {
             worker_tags: Default::default(),
             priority_tags: Default::default(),
             dedicated_worker: Default::default(),
+            dedicated_workers: Default::default(),
             init_bash: Default::default(),
             periodic_script_bash: Default::default(),
             periodic_script_interval_seconds: Default::default(),
@@ -1869,6 +1912,7 @@ pub struct WorkerConfig {
     pub worker_tags: Vec<String>,
     pub priority_tags_sorted: Vec<PriorityTags>,
     pub dedicated_worker: Option<WorkspacedPath>,
+    pub dedicated_workers: Option<Vec<WorkspacedPath>>,
     pub init_bash: Option<String>,
     pub periodic_script_bash: Option<String>,
     pub periodic_script_interval_seconds: Option<u64>,
@@ -1880,8 +1924,8 @@ pub struct WorkerConfig {
 
 impl std::fmt::Debug for WorkerConfig {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?} }}",
-        self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "))
+        write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?} }}",
+        self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "))
     }
 }
 

backend/windmill-indexer/src/completed_runs_oss.rs

@@ -18,7 +18,7 @@ pub struct IndexReader;
 pub struct IndexWriter;
 
 #[cfg(not(feature = "private"))]
-pub async fn init_index(_db: &Pool<Postgres>) -> Result<(IndexReader, IndexWriter), Error> {
+pub async fn init_index(_db: &Pool<Postgres>) -> Result<Option<(IndexReader, IndexWriter)>, Error> {
     Err(anyhow!("Cannot initialize index: not in EE").into())
 }
 

backend/windmill-indexer/src/service_logs_oss.rs

@@ -22,7 +22,7 @@ pub struct ServiceLogIndexWriter;
 pub async fn init_index(
     _db: &Pool<Postgres>,
     mut _killpill_tx: KillpillSender,
-) -> Result<(ServiceLogIndexReader, ServiceLogIndexWriter), Error> {
+) -> Result<Option<(ServiceLogIndexReader, ServiceLogIndexWriter)>, Error> {
     Err(anyhow!("Cannot initialize index: not in EE").into())
 }
 

backend/windmill-worker/src/ai/providers/bedrock.rs

@@ -51,10 +51,9 @@ impl BedrockQueryBuilder {
     ) -> Result<ParsedResponse, Error> {
         let bedrock_client = if !api_key.is_empty() {
             BedrockClient::from_bearer_token(api_key.to_string(), region).await?
-        } else if let (Some(access_key_id), Some(secret_access_key)) = (
-            aws_access_key_id.filter(|s| !s.is_empty()),
-            aws_secret_access_key.filter(|s| !s.is_empty()),
-        ) {
+        } else if let (Some(access_key_id), Some(secret_access_key)) =
+            (aws_access_key_id, aws_secret_access_key)
+        {
             BedrockClient::from_credentials(
                 access_key_id.to_string(),
                 secret_access_key.to_string(),

backend/windmill-worker/src/ai/types.rs

@@ -14,7 +14,11 @@ pub struct McpToolSource {
     pub resource_path: String,
 }
 use windmill_common::{
-    ai_providers::AIProvider, db::DB, error::Error, flow_status::AgentAction, flows::FlowModule,
+    ai_providers::{empty_string_as_none, AIProvider},
+    db::DB,
+    error::Error,
+    flow_status::AgentAction,
+    flows::FlowModule,
     s3_helpers::S3Object,
 };
 use windmill_parser::Typ;
@@ -162,17 +166,18 @@ pub enum AnthropicPlatform {
 
 #[derive(Deserialize, Debug)]
 pub struct ProviderResource {
-    #[serde(alias = "apiKey")]
+    #[serde(alias = "apiKey", default, deserialize_with = "empty_string_as_none")]
     pub api_key: Option<String>,
-    #[serde(alias = "baseUrl")]
+    #[serde(alias = "baseUrl", default, deserialize_with = "empty_string_as_none")]
     pub base_url: Option<String>,
     #[allow(dead_code)]
+    #[serde(default, deserialize_with = "empty_string_as_none")]
     pub region: Option<String>,
     #[allow(dead_code)]
-    #[serde(alias = "awsAccessKeyId")]
+    #[serde(alias = "awsAccessKeyId", default, deserialize_with = "empty_string_as_none")]
     pub aws_access_key_id: Option<String>,
     #[allow(dead_code)]
-    #[serde(alias = "awsSecretAccessKey")]
+    #[serde(alias = "awsSecretAccessKey", default, deserialize_with = "empty_string_as_none")]
     pub aws_secret_access_key: Option<String>,
     /// Platform for Anthropic API (standard or google_vertex_ai)
     #[serde(default)]
@@ -199,7 +204,6 @@ impl ProviderWithResource {
         self.kind
             .get_base_url(
                 self.resource.base_url.clone(),
-                self.resource.region.clone(),
                 db,
             )
             .await

backend/windmill-worker/src/ai_executor.rs

@@ -21,7 +21,7 @@ use windmill_mcp::McpClient;
 #[cfg(not(feature = "mcp"))]
 use crate::ai::tools::McpClientStub as McpClient;
 use windmill_common::{
-    ai_providers::AIProvider,
+    ai_providers::{AIProvider},
     cache,
     client::AuthedClient,
     db::DB,
@@ -410,9 +410,14 @@ pub async fn run_agent(
     has_websearch: bool,
 ) -> error::Result<Box<RawValue>> {
     let output_type = args.output_type.as_ref().unwrap_or(&OutputType::Text);
-    let base_url = args.provider.get_base_url(db).await?;
+    // Skip get_base_url for Bedrock - it uses SDK directly, not HTTP
+    let base_url = if args.provider.kind == AIProvider::AWSBedrock {
+        String::new()
+    } else {
+        args.provider.get_base_url(db).await?
+    };
     let api_key = args.provider.get_api_key().unwrap_or("");
-
+    
     // Create the query builder for the provider
     let query_builder = create_query_builder(&args.provider);
 
@@ -660,12 +665,7 @@ pub async fn run_agent(
         let parsed = if args.provider.kind == AIProvider::AWSBedrock {
             #[cfg(feature = "bedrock")]
             {
-                let region = args.provider.get_region();
-                let Some(region) = region else {
-                    return Err(Error::internal_err(
-                        "AWS Bedrock region is required".to_string(),
-                    ));
-                };
+                let region = args.provider.get_region().unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
                 // Use Bedrock SDK via dedicated query builder
                 crate::ai::providers::bedrock::BedrockQueryBuilder::default()
                     .execute_request(

backend/windmill-worker/src/common.rs

@@ -533,6 +533,7 @@ pub async fn update_worker_ping_for_failed_init_script(
                         ip: None,
                         tags: None,
                         dw: None,
+                        dws: None,
                         jobs_executed: None,
                         occupancy_rate: None,
                         occupancy_rate_15s: None,

backend/windmill-worker/src/worker.rs

@@ -1529,7 +1529,10 @@ pub async fn run_worker(
     let mut occupancy_metrics = OccupancyMetrics::new(start_time);
     let mut jobs_executed = 0;
 
-    let is_dedicated_worker: bool = WORKER_CONFIG.read().await.dedicated_worker.is_some();
+    let is_dedicated_worker: bool = {
+        let config = WORKER_CONFIG.read().await;
+        config.dedicated_worker.is_some() || config.dedicated_workers.as_ref().is_some_and(|dws| !dws.is_empty())
+    };
 
     #[cfg(feature = "benchmark")]
     let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS")
@@ -1631,9 +1634,9 @@ pub async fn run_worker(
     // Option<JoinHandle<()>>,
 
     #[cfg(all(feature = "private", feature = "enterprise"))]
-    let (dedicated_workers, is_flow_worker, dedicated_handles): (
+    let (dedicated_workers, dedicated_flow_paths, dedicated_handles): (
         HashMap<String, Sender<DedicatedWorkerJob>>,
-        bool,
+        HashSet<String>,
         Vec<JoinHandle<()>>,
     ) = match conn {
         Connection::Sql(pool) => {
@@ -1648,15 +1651,15 @@ pub async fn run_worker(
             )
             .await
         }
-        Connection::Http(_) => (HashMap::new(), false, vec![]),
+        Connection::Http(_) => (HashMap::new(), HashSet::new(), vec![]),
     };
 
     #[cfg(any(not(feature = "private"), not(feature = "enterprise")))]
-    let (dedicated_workers, is_flow_worker, dedicated_handles): (
+    let (dedicated_workers, dedicated_flow_paths, dedicated_handles): (
         HashMap<String, Sender<DedicatedWorkerJob>>,
-        bool,
+        HashSet<String>,
         Vec<JoinHandle<()>>,
-    ) = (HashMap::new(), false, vec![]);
+    ) = (HashMap::new(), HashSet::new(), vec![]);
 
     if i_worker == 1 {
         if let Err(e) = queue_init_bash_maybe(conn, same_worker_tx.clone(), &worker_name).await {
@@ -2030,30 +2033,32 @@ pub async fn run_worker(
                     JobKind::Script | JobKind::Preview | JobKind::FlowScript
                 ) {
                     if !dedicated_workers.is_empty() {
-                        let key_o = if is_flow_worker {
-                            job.flow_step_id.as_ref().map(|x| x.to_string())
+                        // Try flow path + step_id combinations for flow jobs, otherwise use runnable_path
+                        let dedicated_worker_tx = if let Some(step_id) = job.flow_step_id.as_ref() {
+                            dedicated_flow_paths.iter().find_map(|flow_path| {
+                                let key = format!("{}:{}", flow_path, step_id);
+                                dedicated_workers.get(&key)
+                            })
                         } else {
-                            job.runnable_path.as_ref().map(|x| x.to_string())
+                            job.runnable_path.as_ref().and_then(|path| dedicated_workers.get(path))
                         };
-                        if let Some(key) = key_o {
-                            if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) {
-                                let dedicated_job = DedicatedWorkerJob {
-                                    job: Arc::new(job.job()),
-                                    flow_runners: None,
-                                    done_tx: None,
-                                };
-                                if let Err(e) = dedicated_worker_tx.send(dedicated_job).await {
-                                    tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
-                                }
-
-                                #[cfg(feature = "benchmark")]
-                                {
-                                    add_time!(bench, "sent to dedicated worker");
-                                    infos.add_iter(bench, true);
-                                }
-
-                                continue;
+                        if let Some(dedicated_worker_tx) = dedicated_worker_tx {
+                            let dedicated_job = DedicatedWorkerJob {
+                                job: Arc::new(job.job()),
+                                flow_runners: None,
+                                done_tx: None,
+                            };
+                            if let Err(e) = dedicated_worker_tx.send(dedicated_job).await {
+                                tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
                             }
+
+                            #[cfg(feature = "benchmark")]
+                            {
+                                add_time!(bench, "sent to dedicated worker");
+                                infos.add_iter(bench, true);
+                            }
+
+                            continue;
                         }
                     }
 

backend/windmill-worker/src/worker_utils.rs

@@ -133,6 +133,7 @@ async fn update_worker_ping_full_inner(
                         ip: None,
                         tags: Some(tags.to_vec()),
                         dw: None,
+                        dws: None,
                         jobs_executed: Some(jobs_executed),
                         occupancy_rate: Some(occupancy_rate),
                         occupancy_rate_15s: Some(occupancy_rate_15s.unwrap_or(0.0)),
@@ -159,13 +160,19 @@ pub async fn insert_ping(
     ip: &str,
     db: &Connection,
 ) -> anyhow::Result<()> {
-    let (tags, dw) = {
+    let (tags, dw, dws) = {
         let wc = WORKER_CONFIG.read().await.clone();
         (
             wc.worker_tags,
             wc.dedicated_worker
                 .as_ref()
                 .map(|x| format!("{}:{}", x.workspace_id, x.path)),
+            wc.dedicated_workers.as_ref().map(|workers| {
+                workers
+                    .iter()
+                    .map(|x| format!("{}:{}", x.workspace_id, x.path))
+                    .collect::<Vec<_>>()
+            }),
         )
     };
 
@@ -190,6 +197,7 @@ pub async fn insert_ping(
                 ip,
                 tags.as_slice(),
                 dw,
+                dws.as_deref(),
                 windmill_common::utils::GIT_VERSION,
                 vcpus,
                 memory,
@@ -210,6 +218,7 @@ pub async fn insert_ping(
                         ip: Some(ip.to_string()),
                         tags: Some(tags.to_vec()),
                         dw: dw,
+                        dws: dws,
                         jobs_executed: None,
                         occupancy_rate: None,
                         occupancy_rate_15s: None,
@@ -282,6 +291,7 @@ pub async fn update_worker_ping_from_job(
                         ip: None,
                         tags: None,
                         dw: None,
+                        dws: None,
                         version: None,
                         vcpus: None,
                         memory: None,

benchmarks/lib.ts

@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
 import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
 import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
 
-export const VERSION = "v1.621.1";
+export const VERSION = "v1.621.2";
 
 export async function login(email: string, password: string): Promise<string> {
   return await windmill.UserService.login({

cli/src/commands/app/dev.ts

@@ -306,39 +306,53 @@ interface DevOptions extends GlobalOptions {
   open?: boolean;
 }
 
-async function dev(opts: DevOptions) {
+async function dev(opts: DevOptions, appFolder?: string) {
   GLOBAL_CONFIG_OPT.noCdToRoot = true;
 
   // Search for wmill.yaml by traversing upward (without git root constraint)
   // to initialize nonDottedPaths setting before using folder suffix functions
   await loadNonDottedPathsSetting();
 
-  // Validate that we're in a .raw_app folder
-  const cwd = process.cwd();
-  const currentDirName = path.basename(cwd);
+  // Resolve target directory from argument or use current directory
+  const originalCwd = process.cwd();
+  let targetDir = originalCwd;
 
-  if (!hasFolderSuffix(currentDirName, "raw_app")) {
+  if (appFolder) {
+    targetDir = path.isAbsolute(appFolder)
+      ? appFolder
+      : path.join(originalCwd, appFolder);
+
+    if (!fs.existsSync(targetDir)) {
+      log.error(colors.red(`Error: Directory not found: ${targetDir}`));
+      Deno.exit(1);
+    }
+  }
+
+  // Validate that target is a .raw_app folder
+  const targetDirName = path.basename(targetDir);
+
+  if (!hasFolderSuffix(targetDirName, "raw_app")) {
     log.error(
       colors.red(
         `Error: The dev command must be run inside a ${
           getFolderSuffix("raw_app")
         } folder.\n` +
-          `Current directory: ${currentDirName}\n` +
+          `Target directory: ${targetDirName}\n` +
           `Please navigate to a folder ending with '${
             getFolderSuffix("raw_app")
-          }' before running this command.`,
+          }' or specify one as argument.`,
       ),
     );
     Deno.exit(1);
   }
 
-  // Check for raw_app.yaml
-  const rawAppPath = path.join(cwd, "raw_app.yaml");
+  // Check for raw_app.yaml in target directory
+  const rawAppPath = path.join(targetDir, "raw_app.yaml");
   if (!fs.existsSync(rawAppPath)) {
     log.error(
       colors.red(
-        `Error: raw_app.yaml not found in current directory.\n` +
-          `The dev command must be run in a ${
+        `Error: raw_app.yaml not found in ${targetDir}.\n` +
+          `The dev command requires a ${
             getFolderSuffix("raw_app")
           } folder containing a raw_app.yaml file.`,
       ),
@@ -346,11 +360,16 @@ async function dev(opts: DevOptions) {
     Deno.exit(1);
   }
 
-  // Resolve workspace and authenticate
+  // Resolve workspace and authenticate (from original cwd to find wmill.yaml)
   const workspace = await resolveWorkspace(opts);
   await requireLogin(opts);
   const workspaceId = workspace.workspaceId;
 
+  // Change to target directory for the rest of the command
+  if (appFolder) {
+    process.chdir(targetDir);
+  }
+
   // Load app path from raw_app.yaml
   const rawApp = (await yamlParseFile(rawAppPath)) as any;
   const appPath = rawApp?.custom_path ?? "u/unknown/newapp";
@@ -1289,6 +1308,7 @@ const command = new Command()
   .description(
     "Start a development server for building apps with live reload and hot module replacement",
   )
+  .arguments("[app_folder:string]")
   .option(
     "--port <port:number>",
     "Port to run the dev server on (will find next available port if occupied)",

cli/src/commands/sync/sync.ts

@@ -133,18 +133,6 @@ export function findCodebase(
     return;
   }
   for (const c of codebases) {
-    // First check if the path is within this codebase's relative_path
-    const codebasePath = c.relative_path.replaceAll("\\", "/");
-    const normalizedPath = path.replaceAll("\\", "/");
-    if (!normalizedPath.startsWith(codebasePath + "/") && normalizedPath !== codebasePath) {
-      continue;
-    }
-
-    // Get the path relative to the codebase root for pattern matching
-    const relativePath = normalizedPath.startsWith(codebasePath + "/")
-      ? normalizedPath.substring(codebasePath.length + 1)
-      : normalizedPath;
-
     let included = false;
     let excluded = false;
     if (c.includes == undefined || c.includes == null) {
@@ -157,7 +145,7 @@ export function findCodebase(
       if (included) {
         break;
       }
-      if (minimatch(relativePath, r)) {
+      if (minimatch(path, r)) {
         included = true;
       }
     }
@@ -165,7 +153,7 @@ export function findCodebase(
       c.excludes = [c.excludes];
     }
     for (const r of c.excludes ?? []) {
-      if (minimatch(relativePath, r)) {
+      if (minimatch(path, r)) {
         excluded = true;
       }
     }

cli/src/main.ts

@@ -77,7 +77,7 @@ export {
 //   }
 // });
 
-export const VERSION = "1.621.1";
+export const VERSION = "1.621.2";
 
 // Re-exported from constants.ts to maintain backwards compatibility
 export { WM_FORK_PREFIX } from "./core/constants.ts";

cli/test/preview.test.ts

@@ -306,6 +306,85 @@ export function main(name: string = "World") {
   sanitizeOps: false,
 });
 
+Deno.test({
+  name: "script preview: codebase with imports (simulates ../shared layout)",
+  async fn() {
+    await withTestBackend(async (backend, tempDir) => {
+      // This test simulates a codebase that could be in a parent directory.
+      // The structure is:
+      //   tempDir/
+      //     wmill.yaml (codebase at ".")
+      //     f/
+      //       lib/
+      //         helper.ts       (shared module)
+      //         main_script.ts  (imports helper)
+      //
+      // This tests that codebase bundling correctly includes imported modules,
+      // which is the key functionality needed for ../shared codebases during sync.
+      // Note: Preview requires valid windmill paths (u/, g/, f/), so we run
+      // from within the codebase directory.
+
+      await createWmillConfig(tempDir, {
+        defaultTs: "bun",
+        codebases: [{ relative_path: ".", includes: ["**"] }],
+      });
+
+      // Create helper module
+      await Deno.mkdir(`${tempDir}/f/lib`, { recursive: true });
+      await Deno.writeTextFile(
+        `${tempDir}/f/lib/helper.ts`,
+        `export function greet(name: string): string {
+  return \`Hello from shared codebase, \${name}!\`;
+}`
+      );
+
+      // Create main script that imports the helper
+      await Deno.writeTextFile(
+        `${tempDir}/f/lib/main_script.ts`,
+        `import { greet } from "./helper";
+
+export function main(name: string = "World") {
+  console.log("Running codebase script with imports");
+  return greet(name);
+}`
+      );
+
+      // Create script metadata
+      await Deno.writeTextFile(
+        `${tempDir}/f/lib/main_script.script.yaml`,
+        `summary: "Test script with imports"
+description: "Test script that imports from helper module"
+lock: ""
+schema:
+  $schema: "https://json-schema.org/draft/2020-12/schema"
+  type: object
+  properties:
+    name:
+      type: string
+      default: "World"
+  required: []
+`
+      );
+
+      // Run preview - the script should be bundled with the helper module
+      const result = await backend.runCLICommand(
+        ["script", "preview", "f/lib/main_script.ts"],
+        tempDir
+      );
+
+      assertEquals(result.code, 0, `Preview failed: ${result.stderr}\n${result.stdout}`);
+      // The script should be bundled (includes the helper) and run successfully
+      assertStringIncludes(
+        result.stdout + result.stderr,
+        "Hello from shared codebase, World!",
+        `Expected codebase script output not found. Got: ${result.stdout}\n${result.stderr}`
+      );
+    });
+  },
+  sanitizeResources: false,
+  sanitizeOps: false,
+});
+
 // =============================================================================
 // FLOW PREVIEW TESTS
 // =============================================================================

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
 	"name": "windmill-components",
-	"version": "1.621.1",
+	"version": "1.621.2",
 	"lockfileVersion": 3,
 	"requires": true,
 	"packages": {
 		"": {
 			"name": "windmill-components",
-			"version": "1.621.1",
+			"version": "1.621.2",
 			"hasInstallScript": true,
 			"license": "AGPL-3.0",
 			"dependencies": {

frontend/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "windmill-components",
-	"version": "1.621.1",
+	"version": "1.621.2",
 	"scripts": {
 		"dev": "vite dev",
 		"build": "vite build",

frontend/src/lib/components/ApiConnectForm.svelte

@@ -14,6 +14,7 @@
 	import { base } from '$lib/base'
 	import GitHubAppIntegration from './GitHubAppIntegration.svelte'
 	import BedrockCredentialsCheck from './BedrockCredentialsCheck.svelte'
+	import { isCloudHosted } from '$lib/cloud'
 
 	interface Props {
 		resourceType: string
@@ -211,7 +212,7 @@
 			onDescriptionUpdate={(newDescription) => (description = newDescription)}
 		/>
 	</div>
-	{#if resourceType?.includes('bedrock')}
+	{#if resourceType?.includes('bedrock') && !isCloudHosted()}
 		<BedrockCredentialsCheck />
 	{/if}
 {:else}

frontend/src/lib/components/DedicatedWorkersSelector.svelte

@@ -0,0 +1,645 @@
+<script lang="ts">
+	import { ScriptService, FlowService, WorkspaceService, type FlowModule } from '$lib/gen'
+	import { Check, X, RefreshCcw, ChevronDown, ChevronRight, CodeXml } from 'lucide-svelte'
+	import { Button } from './common'
+	import Select from './select/Select.svelte'
+	import { sendUserToast } from '$lib/toast'
+	import Badge from './common/badge/Badge.svelte'
+	import { SvelteMap } from 'svelte/reactivity'
+	import { untrack } from 'svelte'
+	import BarsStaggered from './icons/BarsStaggered.svelte'
+
+	// A "Runnable" is a script or flow with dedicated_worker=true
+	interface Runnable {
+		tag: string // workspace:path or workspace:flow/path
+		displayName: string
+		language: string
+		type: 'script' | 'flow'
+		path: string
+		selected: boolean
+		// For flows, the actual runners (steps) that will be spawned
+		runners?: FlowRunner[]
+		loadingRunners?: boolean
+		expanded?: boolean
+	}
+
+	// A "FlowRunner" is an individual step within a flow that will get a dedicated worker
+	interface FlowRunner {
+		stepId: string
+		stepSummary?: string
+		language?: string
+		scriptPath?: string
+		isInline: boolean
+	}
+
+	interface Props {
+		selectedTags: string[]
+		disabled?: boolean
+		onchange?: (tags: string[]) => void
+	}
+
+	let { selectedTags = $bindable([]), disabled = false, onchange }: Props = $props()
+
+	let selectedWorkspace: string | undefined = $state(undefined)
+	let runnables: Runnable[] = $state([])
+	let loading = $state(false)
+	let workspaces: { id: string; name: string }[] = $state([])
+	let workspacesLoading = $state(true)
+	let selectorExpanded = $state(false)
+
+	// Track detailed info for each selected tag (for displaying in summary)
+	interface SelectedTagInfo {
+		tag: string
+		workspace: string
+		type: 'script' | 'flow'
+		path: string
+		runners?: FlowRunner[]
+		expanded?: boolean
+		loading?: boolean
+	}
+	let selectedTagsInfo: SvelteMap<string, SelectedTagInfo> = $state(new SvelteMap())
+
+	// Languages that support dedicated workers
+	const DEDICATED_WORKER_LANGUAGES = ['python3', 'bun', 'deno']
+
+	// Parse a tag to extract workspace, type (script/flow), and path
+	function parseTag(tag: string): { workspace: string; type: 'script' | 'flow'; path: string } | null {
+		const colonIndex = tag.indexOf(':')
+		if (colonIndex === -1) return null
+
+		const workspace = tag.substring(0, colonIndex)
+		const rest = tag.substring(colonIndex + 1)
+
+		if (rest.startsWith('flow/')) {
+			return { workspace, type: 'flow', path: rest.substring(5) }
+		} else {
+			return { workspace, type: 'script', path: rest }
+		}
+	}
+
+	// Resolve workspace script languages and filter to supported languages
+	async function resolveAndFilterRunners(
+		workspace: string,
+		preliminaryRunners: FlowRunner[]
+	): Promise<FlowRunner[]> {
+		const runnersWithLanguage = await Promise.all(
+			preliminaryRunners.map(async (runner) => {
+				if (!runner.isInline && runner.scriptPath) {
+					try {
+						const script = await ScriptService.getScriptByPath({
+							workspace,
+							path: runner.scriptPath
+						})
+						return { ...runner, language: script.language }
+					} catch (e) {
+						console.error(`Failed to fetch script ${runner.scriptPath}`, e)
+						return { ...runner, language: undefined }
+					}
+				}
+				return runner
+			})
+		)
+
+		// Filter to only supported languages
+		return runnersWithLanguage.filter(
+			(runner) => runner.language && DEDICATED_WORKER_LANGUAGES.includes(runner.language)
+		)
+	}
+
+	// Load detailed info for all selected tags
+	async function loadSelectedTagsInfo(tags: string[]) {
+		if (tags.length === 0) {
+			selectedTagsInfo = new SvelteMap()
+			return
+		}
+
+		// Capture current state without tracking to avoid infinite loops
+		const currentInfo = untrack(() => selectedTagsInfo)
+		const currentRunnables = untrack(() => runnables)
+
+		const newInfo = new SvelteMap<string, SelectedTagInfo>()
+
+		try {
+			await Promise.all(
+				tags.map(async (tag) => {
+					// Check if we already have this info cached
+					const existing = currentInfo.get(tag)
+					if (existing && (existing.type === 'script' || existing.runners !== undefined)) {
+						newInfo.set(tag, existing)
+						return
+					}
+
+					// Check if we have it loaded in runnables
+					const existingRunnable = currentRunnables.find((r) => r.tag === tag)
+					if (existingRunnable) {
+						newInfo.set(tag, {
+							tag,
+							workspace: tag.substring(0, tag.indexOf(':')),
+							type: existingRunnable.type,
+							path: existingRunnable.path,
+							runners: existingRunnable.runners,
+							expanded: existing?.expanded ?? false
+						})
+						return
+					}
+
+					// Parse and fetch
+					const parsed = parseTag(tag)
+					if (!parsed) return
+
+					if (parsed.type === 'script') {
+						newInfo.set(tag, {
+							tag,
+							workspace: parsed.workspace,
+							type: 'script',
+							path: parsed.path
+						})
+					} else {
+						// Flows need to fetch to get runners
+						try {
+							const flow = await FlowService.getFlowByPath({
+								workspace: parsed.workspace,
+								path: parsed.path
+							})
+							const preliminaryRunners = flow.value?.modules
+								? extractRunnersFromModules(flow.value.modules)
+								: []
+							const runners = await resolveAndFilterRunners(parsed.workspace, preliminaryRunners)
+							newInfo.set(tag, {
+								tag,
+								workspace: parsed.workspace,
+								type: 'flow',
+								path: parsed.path,
+								runners,
+								expanded: existing?.expanded ?? false
+							})
+						} catch (e) {
+							console.error(`Failed to load flow ${parsed.path}`, e)
+							newInfo.set(tag, {
+								tag,
+								workspace: parsed.workspace,
+								type: 'flow',
+								path: parsed.path,
+								runners: []
+							})
+						}
+					}
+				})
+			)
+		} finally {
+			selectedTagsInfo = newInfo
+		}
+	}
+
+	function toggleSelectedTagExpanded(tag: string) {
+		const info = selectedTagsInfo.get(tag)
+		if (info) {
+			// Need to set the whole object to trigger reactivity
+			selectedTagsInfo.set(tag, { ...info, expanded: !info.expanded })
+		}
+	}
+
+	// Auto-expand selector if no tags selected
+	$effect(() => {
+		if (selectedTags.length === 0) {
+			selectorExpanded = true
+		}
+	})
+
+	// Load selected tags info when selectedTags change
+	$effect(() => {
+		loadSelectedTagsInfo(selectedTags)
+	})
+
+	$effect(() => {
+		loadWorkspaces()
+	})
+
+	async function loadWorkspaces() {
+		try {
+			workspacesLoading = true
+			const ws = await WorkspaceService.listWorkspaces()
+			workspaces = ws.map((w) => ({ id: w.id, name: w.name }))
+		} catch (e) {
+			console.error('Failed to load workspaces', e)
+			sendUserToast('Failed to load workspaces', true)
+		} finally {
+			workspacesLoading = false
+		}
+	}
+
+	// Extract runners from flow modules recursively
+	// Returns runners with language info for inline scripts, and scriptPath for workspace scripts
+	function extractRunnersFromModules(modules: FlowModule[]): FlowRunner[] {
+		const runners: FlowRunner[] = []
+
+		for (const module of modules) {
+			const value = module.value
+			switch (value.type) {
+				case 'rawscript':
+					if (DEDICATED_WORKER_LANGUAGES.includes(value.language)) {
+						runners.push({
+							stepId: module.id,
+							stepSummary: module.summary,
+							language: value.language,
+							scriptPath: value.path,
+							isInline: true
+						})
+					}
+					break
+				case 'script':
+					// For workspace script references, we'll resolve the language later
+					runners.push({
+						stepId: module.id,
+						stepSummary: module.summary,
+						language: undefined, // Will be resolved by fetching the script
+						scriptPath: value.path,
+						isInline: false
+					})
+					break
+				case 'forloopflow':
+					runners.push(...extractRunnersFromModules(value.modules))
+					break
+				case 'whileloopflow':
+					runners.push(...extractRunnersFromModules(value.modules))
+					break
+				case 'branchone':
+					for (const branch of value.branches) {
+						runners.push(...extractRunnersFromModules(branch.modules))
+					}
+					runners.push(...extractRunnersFromModules(value.default))
+					break
+				case 'branchall':
+					for (const branch of value.branches) {
+						runners.push(...extractRunnersFromModules(branch.modules))
+					}
+					break
+			}
+		}
+
+		return runners
+	}
+
+	async function loadFlowRunners(runnable: Runnable) {
+		if (!selectedWorkspace || runnable.type !== 'flow') return
+
+		try {
+			runnable.loadingRunners = true
+			const flow = await FlowService.getFlowByPath({
+				workspace: selectedWorkspace,
+				path: runnable.path
+			})
+
+			if (flow.value?.modules) {
+				const preliminaryRunners = extractRunnersFromModules(flow.value.modules)
+				runnable.runners = await resolveAndFilterRunners(selectedWorkspace, preliminaryRunners)
+			} else {
+				runnable.runners = []
+			}
+		} catch (e) {
+			console.error('Failed to load flow runners', e)
+			runnable.runners = []
+		} finally {
+			runnable.loadingRunners = false
+		}
+	}
+
+	async function loadRunnables(workspaceId: string) {
+		try {
+			loading = true
+			runnables = []
+
+			const [scripts, flows] = await Promise.all([
+				ScriptService.listScripts({
+					workspace: workspaceId,
+					dedicatedWorker: true
+				}),
+				FlowService.listFlows({
+					workspace: workspaceId,
+					dedicatedWorker: true
+				})
+			])
+
+			const newRunnables: Runnable[] = []
+
+			// Add scripts with supported languages
+			for (const script of scripts) {
+				if (DEDICATED_WORKER_LANGUAGES.includes(script.language ?? '')) {
+					const tag = `${workspaceId}:${script.path}`
+					newRunnables.push({
+						tag,
+						displayName: script.path,
+						language: script.language ?? 'unknown',
+						type: 'script',
+						path: script.path,
+						selected: selectedTags.includes(tag)
+					})
+				}
+			}
+
+			// Add flows
+			for (const flow of flows) {
+				const tag = `${workspaceId}:flow/${flow.path}`
+				newRunnables.push({
+					tag,
+					displayName: flow.path,
+					language: 'flow',
+					type: 'flow',
+					path: flow.path,
+					selected: selectedTags.includes(tag),
+					runners: undefined,
+					loadingRunners: false,
+					expanded: false
+				})
+			}
+
+			runnables = newRunnables
+
+			// Load runners for all flows in parallel
+			await Promise.all(runnables.filter((r) => r.type === 'flow').map((r) => loadFlowRunners(r)))
+		} catch (e) {
+			console.error('Failed to load runnables', e)
+			sendUserToast('Failed to load scripts/flows', true)
+		} finally {
+			loading = false
+		}
+	}
+
+	function toggleRunnable(runnable: Runnable) {
+		runnable.selected = !runnable.selected
+		updateSelectedTags()
+	}
+
+	function toggleExpanded(runnable: Runnable) {
+		runnable.expanded = !runnable.expanded
+	}
+
+	function selectAll() {
+		for (const runnable of runnables) {
+			runnable.selected = true
+		}
+		updateSelectedTags()
+	}
+
+	function deselectAll() {
+		for (const runnable of runnables) {
+			runnable.selected = false
+		}
+		updateSelectedTags()
+	}
+
+	function updateSelectedTags() {
+		selectedTags = runnables.filter((r) => r.selected).map((r) => r.tag)
+		onchange?.(selectedTags)
+	}
+
+	function removeTag(tag: string) {
+		selectedTags = selectedTags.filter((t) => t !== tag)
+		// Also update runnable state if visible
+		const runnable = runnables.find((r) => r.tag === tag)
+		if (runnable) {
+			runnable.selected = false
+		}
+		onchange?.(selectedTags)
+	}
+
+	$effect(() => {
+		if (selectedWorkspace) {
+			loadRunnables(selectedWorkspace)
+		}
+	})
+
+	let selectedCount = $derived(runnables.filter((r) => r.selected).length)
+</script>
+
+<div class="flex flex-col gap-3">
+	<!-- Selected tags summary -->
+	{#if selectedTags.length > 0}
+		<div class="flex flex-col gap-2">
+			<div class="border rounded-md divide-y bg-surface max-h-48 overflow-y-auto">
+				{#each selectedTags as tag (tag)}
+					{@const info = selectedTagsInfo.get(tag)}
+					<div>
+						<div class="flex items-center">
+							{#if info?.type === 'flow' && info.runners && info.runners.length > 0}
+								<button
+									class="p-2 hover:bg-surface-hover transition-colors"
+									onclick={(e) => {
+										e.stopPropagation()
+										toggleSelectedTagExpanded(tag)
+									}}
+								>
+									{#if info.expanded}
+										<ChevronDown class="h-3 w-3 text-tertiary" />
+									{:else}
+										<ChevronRight class="h-3 w-3 text-tertiary" />
+									{/if}
+								</button>
+							{:else}
+								<div class="w-7"></div>
+							{/if}
+							<div class="flex-1 flex items-center gap-2 px-2 py-1.5 min-w-0">
+								{#if info}
+									{#if info.type === 'flow'}
+										<BarsStaggered size={14} class="flex-shrink-0 text-secondary" />
+									{:else}
+										<CodeXml size={14} class="flex-shrink-0 text-secondary" />
+									{/if}
+									<span class="text-xs truncate flex-1">{info.path}</span>
+									<span class="text-xs text-tertiary flex-shrink-0">({info.workspace})</span>
+									{#if info.type === 'flow' && info.runners}
+										<Badge color="indigo" small>
+											{info.runners.length} runner{info.runners.length !== 1 ? 's' : ''}
+										</Badge>
+									{:else if info.type === 'script'}
+										<Badge color="blue" small>1 runner</Badge>
+									{/if}
+								{:else}
+									<span class="text-xs text-tertiary truncate">{tag}</span>
+								{/if}
+							</div>
+							{#if !disabled}
+								<button
+									class="p-2 hover:text-red-500 transition-colors"
+									onclick={(e) => {
+										e.stopPropagation()
+										removeTag(tag)
+									}}
+								>
+									<X class="h-3 w-3" />
+								</button>
+							{/if}
+						</div>
+
+						{#if info?.type === 'flow' && info.expanded && info.runners}
+							<div class="bg-surface-secondary border-t">
+								{#each info.runners as runner (runner.stepId)}
+									<div class="flex items-center gap-2 px-9 py-1 text-xs border-t first:border-t-0">
+										<span class="font-mono text-tertiary">{runner.stepId}</span>
+										{#if runner.stepSummary}
+											<span class="text-secondary truncate flex-1">{runner.stepSummary}</span>
+										{/if}
+										<Badge color="gray" small>
+											{runner.isInline ? runner.language : runner.scriptPath}
+										</Badge>
+									</div>
+								{/each}
+							</div>
+						{/if}
+					</div>
+				{/each}
+			</div>
+		</div>
+	{/if}
+
+	<!-- Collapsible selector section -->
+	<div class="border rounded-md">
+		<button
+			class="w-full flex items-center gap-2 px-3 py-2 text-left hover:bg-surface-hover transition-colors"
+			onclick={() => (selectorExpanded = !selectorExpanded)}
+			{disabled}
+		>
+			{#if selectorExpanded}
+				<ChevronDown class="h-4 w-4 text-secondary" />
+			{:else}
+				<ChevronRight class="h-4 w-4 text-secondary" />
+			{/if}
+			<span class="text-sm">
+				{selectedTags.length > 0 ? 'Add more scripts/flows' : 'Select scripts/flows'}
+			</span>
+		</button>
+
+		{#if selectorExpanded}
+			<div class="border-t px-3 py-3 flex flex-col gap-3">
+				<!-- Workspace selector -->
+				<div class="flex flex-col gap-1">
+					<span class="text-xs text-secondary">Workspace</span>
+					<Select
+						bind:value={selectedWorkspace}
+						items={workspaces.map((w) => ({ value: w.id, label: `${w.name} (${w.id})` }))}
+						placeholder="Select workspace..."
+						disabled={disabled || workspacesLoading}
+					/>
+				</div>
+
+				<!-- Scripts/flows list -->
+				{#if selectedWorkspace}
+					<div class="flex flex-col gap-2">
+						<div class="flex items-center justify-between">
+							<span class="text-xs text-secondary">Scripts/flows with dedicated worker enabled</span
+							>
+							{#if !loading && runnables.length > 0}
+								<div class="flex gap-1">
+									<Button size="xs2" color="light" on:click={selectAll} {disabled}>All</Button>
+									<Button size="xs2" color="light" on:click={deselectAll} {disabled}>None</Button>
+									<Button
+										size="xs2"
+										color="light"
+										iconOnly
+										startIcon={{ icon: RefreshCcw }}
+										on:click={() => selectedWorkspace && loadRunnables(selectedWorkspace)}
+										{disabled}
+									/>
+								</div>
+							{/if}
+						</div>
+
+						{#if loading}
+							<div class="flex items-center justify-center py-4">
+								<RefreshCcw class="animate-spin h-4 w-4 text-secondary" />
+								<span class="ml-2 text-xs text-secondary">Loading...</span>
+							</div>
+						{:else if runnables.length === 0}
+							<div class="text-xs text-tertiary py-3 text-center">
+								No scripts or flows with dedicated worker enabled found.
+							</div>
+						{:else}
+							<div class="border rounded-md divide-y max-h-64 overflow-y-auto bg-surface">
+								{#each runnables as runnable (runnable.tag)}
+									<div>
+										<div class="flex items-center">
+											{#if runnable.type === 'flow' && runnable.runners && runnable.runners.length > 0}
+												<button
+													class="p-2 hover:bg-surface-hover transition-colors"
+													onclick={(e) => {
+														e.stopPropagation()
+														toggleExpanded(runnable)
+													}}
+													{disabled}
+												>
+													{#if runnable.expanded}
+														<ChevronDown class="h-3 w-3 text-tertiary" />
+													{:else}
+														<ChevronRight class="h-3 w-3 text-tertiary" />
+													{/if}
+												</button>
+											{:else}
+												<div class="w-7"></div>
+											{/if}
+											<button
+												class="flex-1 flex items-center gap-2 px-2 py-1.5 hover:bg-surface-hover transition-colors text-left"
+												onclick={(e) => {
+													e.stopPropagation()
+													if (!disabled) toggleRunnable(runnable)
+												}}
+												{disabled}
+											>
+												<div
+													class="w-4 h-4 border rounded flex items-center justify-center flex-shrink-0"
+													class:bg-blue-500={runnable.selected}
+													class:border-blue-500={runnable.selected}
+												>
+													{#if runnable.selected}
+														<Check class="h-3 w-3 text-white" />
+													{/if}
+												</div>
+												<span class="flex-1 text-xs truncate">{runnable.displayName}</span>
+												{#if runnable.type === 'flow' && runnable.runners}
+													<span class="text-xs text-tertiary">
+														{runnable.runners.length}
+													</span>
+												{/if}
+												<Badge color={runnable.type === 'flow' ? 'indigo' : 'blue'} small>
+													{runnable.type === 'flow' ? 'flow' : runnable.language}
+												</Badge>
+											</button>
+										</div>
+
+										{#if runnable.type === 'flow' && runnable.expanded && runnable.runners}
+											<div class="bg-surface-secondary border-t">
+												{#if runnable.runners.length === 0}
+													<div class="px-9 py-1.5 text-xs text-tertiary italic">
+														No eligible steps (python3/bun/deno)
+													</div>
+												{:else}
+													{#each runnable.runners as runner (runner.stepId)}
+														<div
+															class="flex items-center gap-2 px-9 py-1 text-xs border-t first:border-t-0"
+														>
+															<span class="font-mono text-tertiary">{runner.stepId}</span>
+															{#if runner.stepSummary}
+																<span class="text-secondary truncate flex-1">
+																	{runner.stepSummary}
+																</span>
+															{/if}
+															<Badge color="gray" small>
+																{runner.isInline ? runner.language : runner.scriptPath}
+															</Badge>
+														</div>
+													{/each}
+												{/if}
+											</div>
+										{/if}
+									</div>
+								{/each}
+							</div>
+							<div class="text-xs text-tertiary">
+								{selectedCount} selected
+							</div>
+						{/if}
+					</div>
+				{/if}
+			</div>
+		{/if}
+	</div>
+</div>

frontend/src/lib/components/WorkerGroup.svelte

@@ -42,6 +42,7 @@
 	import TextInput from './text_input/TextInput.svelte'
 	import Dropdown from './DropdownV2.svelte'
 	import TagList from './TagList.svelte'
+	import DedicatedWorkersSelector from './DedicatedWorkersSelector.svelte'
 
 	function computeVCpuAndMemory(workers: [string, WorkerPing[]][]) {
 		let vcpus = 0
@@ -73,6 +74,7 @@
 
 	let nconfig: {
 		dedicated_worker?: string
+		dedicated_workers?: string[]
 		worker_tags?: string[]
 		priority_tags?: Record<string, number>
 		cache_clear?: number
@@ -89,7 +91,9 @@
 
 	function loadNConfig() {
 		nconfig = config
-			? config.worker_tags != undefined || config.dedicated_worker != undefined
+			? config.worker_tags != undefined ||
+				config.dedicated_worker != undefined ||
+				config.dedicated_workers != undefined
 				? config
 				: {
 						worker_tags: []
@@ -101,6 +105,12 @@
 			nconfig.priority_tags = {}
 		}
 
+		// Convert legacy dedicated_worker to dedicated_workers array
+		if (nconfig.dedicated_worker && !nconfig.dedicated_workers?.length) {
+			nconfig.dedicated_workers = [nconfig.dedicated_worker]
+			nconfig.dedicated_worker = undefined
+		}
+
 		customEnvVars = []
 		if (nconfig.env_vars_allowlist === undefined) {
 			nconfig.env_vars_allowlist = []
@@ -170,6 +180,7 @@
 			| undefined
 			| {
 					dedicated_worker?: string
+					dedicated_workers?: string[]
 					worker_tags?: string[]
 					priority_tags?: Record<string, number>
 					cache_clear?: number
@@ -243,7 +254,11 @@
 
 	let drawer: Drawer | undefined = $state()
 	let vcpus_memory = $derived(computeVCpuAndMemory(workers))
-	let selected = $derived(nconfig?.dedicated_worker != undefined ? 'dedicated' : 'normal')
+	let selected = $derived(
+		nconfig?.dedicated_worker != undefined || (nconfig?.dedicated_workers?.length ?? 0) > 0
+			? 'dedicated'
+			: 'normal'
+	)
 	$effect(() => {
 		;($superadmin || $devopsRole) && listWorkspaces()
 	})
@@ -330,17 +345,19 @@
 						nconfig = {}
 					}
 					if (e.detail == 'dedicated') {
-						nconfig.dedicated_worker = ''
+						nconfig.dedicated_workers = nconfig.dedicated_workers ?? []
+						nconfig.dedicated_worker = undefined
 						nconfig.worker_tags = undefined
 					} else {
 						nconfig.dedicated_worker = undefined
+						nconfig.dedicated_workers = undefined
 						nconfig.worker_tags = []
 					}
 				}}
 			>
 				{#snippet children({ item })}
 					<ToggleButton value="normal" label="Any jobs within worker tags" {item} />
-					<ToggleButton value="dedicated" label="Dedicated to a script/flow" {item} />
+					<ToggleButton value="dedicated" label="Dedicated to scripts/flows" {item} />
 				{/snippet}
 			</ToggleButtonGroup>
 		</Label>
@@ -493,30 +510,33 @@
 				</Label>
 			{/if}
 		{:else if selected == 'dedicated'}
-			<div class="flex flex-col gap-2">
+			<div class="flex flex-col gap-4">
 				{#if $superadmin || $devopsRole}
 					<div class="py-2">
 						<Alert
 							size="xs"
 							type="info"
-							title="Script's runtime setting 'dedicated worker' must be toggled on as well"
+							title="The 'dedicated worker' runtime setting of the runnables must be enabled to be selected here"
 						/>
 					</div>
 				{/if}
-				{#if nconfig?.dedicated_worker != undefined}
-					<div
-						><p class="text-xs text-secondary mb-2"
-							>Workers will get killed upon detecting changes. It is assumed they are in an
-							environment where the supervisor will restart them.</p
-						>
-						<input
-							disabled={!canEditConfig}
-							placeholder="<workspace>:<script path>"
-							type="text"
-							onchange={() => {}}
-							bind:value={nconfig.dedicated_worker}
-						/></div
-					>
+
+				<p class="text-xs text-secondary"
+					>Workers will get killed upon detecting changes. It is assumed they are in an environment
+					where the supervisor will restart them.</p
+				>
+
+				{#if nconfig !== undefined}
+					<DedicatedWorkersSelector
+						selectedTags={nconfig.dedicated_workers ?? []}
+						disabled={!canEditConfig}
+						onchange={(tags) => {
+							if (nconfig) {
+								nconfig.dedicated_workers = tags
+								nconfig.dedicated_worker = undefined
+							}
+						}}
+					/>
 				{/if}
 			</div>
 		{/if}

lsp/Pipfile

@@ -4,8 +4,8 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-wmill = ">=1.621.1"
-wmill_pg = ">=1.621.1"
+wmill = ">=1.621.2"
+wmill_pg = ">=1.621.2"
 sendgrid = "*"
 mysql-connector-python = "*"
 pymongo = "*"

openflow.openapi.yaml

@@ -1,7 +1,7 @@
 openapi: '3.0.3'
 
 info:
-  version: 1.621.1
+  version: 1.621.2
   title: OpenFlow Spec
   contact:
     name: Ruben Fiszel

powershell-client/WindmillClient/WindmillClient.psd1

@@ -12,7 +12,7 @@
     RootModule        = 'WindmillClient.psm1'
 
     # Version number of this module.
-    ModuleVersion     = '1.621.1'
+    ModuleVersion     = '1.621.2'
 
     # Supported PSEditions
     # CompatiblePSEditions = @()

python-client/wmill/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill"
-version = "1.621.1"
+version = "1.621.2"
 description = "A client library for accessing Windmill server wrapping the Windmill client API"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"

python-client/wmill_pg/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill-pg"
-version = "1.621.1"
+version = "1.621.2"
 description = "An extension client for the wmill client library focused on pg"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"

typescript-client/jsr.json

@@ -1,6 +1,6 @@
 {
   "name": "@windmill/windmill",
-  "version": "1.621.1",
+  "version": "1.621.2",
   "exports": "./src/index.ts",
   "publish": {
     "exclude": ["!src", "./s3Types.ts", "./sqlUtils.ts", "./client.ts"]

typescript-client/package.json

@@ -1,7 +1,7 @@
 {
   "name": "windmill-client",
   "description": "Windmill SDK client for browsers and Node.js",
-  "version": "1.621.1",
+  "version": "1.621.2",
   "author": "Ruben Fiszel",
   "license": "Apache 2.0",
   "sideEffects": false,

version.txt

@@ -1 +1 @@
-1.621.1
+1.621.2