perf: optimize batch pull query with ANY(ARRAY()) and chained CTEs

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.github/workflows/backend-test.yml

@@ -238,4 +238,4 @@ jobs:
         run: |
           deno --version && bun -v && node --version && go version && python3 --version && php --version && ruby --version && pwsh --version && dotnet --version
           cd windmill-duckdb-ffi-internal && ./build_dev.sh && cd ..
-          DENO_PATH=$(which deno) BUN_PATH=$(which bun) NODE_BIN_PATH=$(which node) GO_PATH=$(which go) UV_PATH=$(which uv) PHP_PATH=$(which php) COMPOSER_PATH=$(which composer) RUBY_PATH=$(which ruby) RUBY_BUNDLE_PATH=$(which bundle) RUBY_GEM_PATH=$(which gem) POWERSHELL_PATH=$(which pwsh) DOTNET_PATH=$(which dotnet) cargo test --features enterprise,deno_core,duckdb,license,python,rust,scoped_cache,parquet,private,private_registry_test,csharp,php,ruby,mysql,quickjs,mcp --all -- --nocapture --test-threads=10
+          DENO_PATH=$(which deno) BUN_PATH=$(which bun) NODE_BIN_PATH=$(which node) GO_PATH=$(which go) UV_PATH=$(which uv) PHP_PATH=$(which php) COMPOSER_PATH=$(which composer) RUBY_PATH=$(which ruby) RUBY_BUNDLE_PATH=$(which bundle) RUBY_GEM_PATH=$(which gem) POWERSHELL_PATH=$(which pwsh) DOTNET_PATH=$(which dotnet) cargo test --features enterprise,deno_core,duckdb,license,python,rust,scoped_cache,parquet,private,private_registry_test,csharp,php,ruby,mysql,quickjs,mcp,run_inline --all -- --nocapture --test-threads=10

.github/workflows/discord-notification.yml

@@ -9,9 +9,7 @@ on:
   issue_comment:
     types:
       - created
-  pull_request_review_comment:
-    types:
-      - created
+      - edited
 
 jobs:
   notify_discord_when_pr_opened:
@@ -53,23 +51,7 @@ jobs:
       COMMENT_BODY: ${{ github.event.comment.body }}
       COMMENT_AUTHOR: ${{ github.event.comment.user.login }}
       COMMENT_URL: ${{ github.event.comment.html_url }}
-      DISCORD_CHANNEL_ID: "1372204995868491786"
-      DISCORD_GUILD_ID: "930051556043276338"
-    secrets:
-      DISCORD_BOT_TOKEN: ${{ secrets.DISCORD_AI_BOT_TOKEN }}
-
-  notify_discord_on_review_comment:
-    if: >
-      github.event_name == 'pull_request_review_comment'
-      && github.event.comment.user.login != 'cloudflare-workers-and-pages[bot]'
-      && github.event.comment.user.login != 'ellipsis-dev[bot]'
-    uses: ./.github/workflows/shareable-discord-notification.yml
-    with:
-      PR_STATUS: "comment"
-      PR_NUMBER: ${{ github.event.pull_request.number }}
-      COMMENT_BODY: ${{ github.event.comment.body }}
-      COMMENT_AUTHOR: ${{ github.event.comment.user.login }}
-      COMMENT_URL: ${{ github.event.comment.html_url }}
+      COMMENT_IS_EDIT: ${{ github.event.action == 'edited' }}
       DISCORD_CHANNEL_ID: "1372204995868491786"
       DISCORD_GUILD_ID: "930051556043276338"
     secrets:

.github/workflows/shareable-discord-notification.yml

@@ -36,6 +36,10 @@ on:
         description: "The comment URL"
         type: string
         default: ""
+      COMMENT_IS_EDIT:
+        description: "Whether this is an edit of an existing comment"
+        type: string
+        default: "false"
     secrets:
       DISCORD_WEBHOOK_URL:
         description: "Discord Webhook URL"
@@ -135,7 +139,7 @@ jobs:
     runs-on: ubuntu-latest
     if: ${{ inputs.PR_STATUS == 'comment' }}
     steps:
-      - name: Post comment to Discord thread
+      - name: Post or update comment in Discord thread
         env:
           BOT_TOKEN: ${{ secrets.DISCORD_BOT_TOKEN }}
           CHANNEL_ID: ${{ inputs.DISCORD_CHANNEL_ID }}
@@ -144,6 +148,7 @@ jobs:
           COMMENT_BODY: ${{ inputs.COMMENT_BODY }}
           COMMENT_AUTHOR: ${{ inputs.COMMENT_AUTHOR }}
           COMMENT_URL: ${{ inputs.COMMENT_URL }}
+          COMMENT_IS_EDIT: ${{ inputs.COMMENT_IS_EDIT }}
         run: |
           # 1) Find the thread by PR number
           threads=$(curl -s -H "Authorization: Bot $BOT_TOKEN" \
@@ -172,10 +177,36 @@ jobs:
             truncated_body="$COMMENT_BODY"
           fi
 
-          # 3) Post the comment to the thread
-          message=$(printf '**%s** [commented](%s):\n%s' "$COMMENT_AUTHOR" "$COMMENT_URL" "$truncated_body")
+          # 3) Build the message content
+          if [ "$COMMENT_IS_EDIT" = "true" ]; then
+            message=$(printf '**%s** [edited comment](%s):\n%s' "$COMMENT_AUTHOR" "$COMMENT_URL" "$truncated_body")
+          else
+            message=$(printf '**%s** [commented](%s):\n%s' "$COMMENT_AUTHOR" "$COMMENT_URL" "$truncated_body")
+          fi
           payload=$(jq -n --arg content "$message" '{content: $content, flags: 4, allowed_mentions: {parse: []}}')
 
+          # 4) If this is an edit, try to find and update the existing Discord message
+          if [ "$COMMENT_IS_EDIT" = "true" ]; then
+            # Search recent messages in the thread for one containing the comment URL
+            messages=$(curl -s -H "Authorization: Bot $BOT_TOKEN" \
+              "https://discord.com/api/v10/channels/${thread_id}/messages?limit=100")
+            existing_msg_id=$(echo "$messages" | jq -r \
+              --arg url "$COMMENT_URL" \
+              '[.[] | select(.content | contains($url))] | first | .id // empty')
+
+            if [ -n "$existing_msg_id" ]; then
+              echo "Updating existing Discord message $existing_msg_id"
+              curl -s -X PATCH \
+                -H "Authorization: Bot $BOT_TOKEN" \
+                -H "Content-Type: application/json" \
+                -d "$payload" \
+                "https://discord.com/api/v10/channels/${thread_id}/messages/${existing_msg_id}"
+              exit 0
+            fi
+            echo "Original Discord message not found, posting as new message"
+          fi
+
+          # 5) Post a new message to the thread
           curl -s -X POST \
             -H "Authorization: Bot $BOT_TOKEN" \
             -H "Content-Type: application/json" \

backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json

@@ -1,202 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT * FROM workspace_settings WHERE teams_team_id = $1 AND teams_command_script IS NOT NULL",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "workspace_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "slack_team_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 2,
-        "name": "slack_name",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "slack_command_script",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 4,
-        "name": "slack_email",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 5,
-        "name": "customer_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 6,
-        "name": "plan",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 7,
-        "name": "webhook",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 8,
-        "name": "deploy_to",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 9,
-        "name": "ai_config",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 10,
-        "name": "large_file_storage",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 11,
-        "name": "git_sync",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 12,
-        "name": "default_app",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 13,
-        "name": "default_scripts",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 14,
-        "name": "deploy_ui",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 15,
-        "name": "mute_critical_alerts",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 16,
-        "name": "color",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 17,
-        "name": "operator_settings",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 18,
-        "name": "teams_command_script",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 19,
-        "name": "teams_team_id",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 20,
-        "name": "teams_team_name",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 21,
-        "name": "git_app_installations",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 22,
-        "name": "ducklake",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 23,
-        "name": "slack_oauth_client_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 24,
-        "name": "slack_oauth_client_secret",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 25,
-        "name": "datatable",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 26,
-        "name": "teams_team_guid",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 27,
-        "name": "auto_invite",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 28,
-        "name": "error_handler",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 29,
-        "name": "success_handler",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 30,
-        "name": "public_app_execution_limit_per_minute",
-        "type_info": "Int4"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      false,
-      true,
-      true,
-      true,
-      false,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      false,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true
-    ]
-  },
-  "hash": "08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7"
-}

backend/.sqlx/query-34721bce20aa8b2a2c6b9bd5455735f1a2270f23d73de95101e6350f6df40acc.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id, teams_command_script FROM workspace_settings WHERE teams_team_id = $1 AND teams_command_script IS NOT NULL",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "teams_command_script",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true
+    ]
+  },
+  "hash": "34721bce20aa8b2a2c6b9bd5455735f1a2270f23d73de95101e6350f6df40acc"
+}

backend/Cargo.lock

@@ -16167,6 +16167,7 @@ dependencies = [
  "windmill-common",
  "windmill-native-triggers",
  "windmill-test-utils",
+ "windmill-worker",
 ]
 
 [[package]]

backend/Cargo.toml

@@ -159,12 +159,12 @@ all_languages = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "
 # For windows we have another set of languages enabled
 all_languages_windows = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql-winauth", "bigquery", "csharp", "nu", "php", "java"]
 # Edition meta-features: shared groups
-inline_preview = ["windmill-api/inline_preview"]
+run_inline = ["windmill-api/run_inline"]
 oss_core = [
     "embedding", "parquet", "openidconnect", "license",
     "http_trigger", "zip", "oauth2", "postgres_trigger",
     "mqtt_trigger", "websocket", "smtp", "native_trigger",
-    "static_frontend", "mcp", "bedrock", "inline_preview",
+    "static_frontend", "mcp", "bedrock", "run_inline",
     "quickjs"
 ]
 ce_core = ["oss_core", "private", "operator"]

backend/ee-repo-ref.txt

@@ -1 +1 @@
-a797dd4d619cdab737e133ce593f2f8582ba21de
+8ffae1f43b31dc8136714fa612d22b6301773e27

backend/src/main.rs

@@ -1822,11 +1822,53 @@ pub async fn run_workers(
             .expect("could not create initial worker dir");
     }
 
+    let native_mode = NATIVE_MODE_RESOLVED.load(std::sync::atomic::Ordering::Relaxed);
+
     tracing::info!(
-        "Starting {num_workers} workers and SLEEP_QUEUE={}ms",
+        "Starting {num_workers} workers (native_polling={native_mode}) and SLEEP_QUEUE={}ms",
         *windmill_worker::SLEEP_QUEUE
     );
 
+    let native_job_rx = if native_mode && num_workers > 1 {
+        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(num_workers));
+        let (job_tx, job_rx) =
+            tokio::sync::mpsc::channel::<windmill_worker::NativeJobDispatch>(num_workers);
+        let native_job_rx: windmill_worker::NativeJobRx =
+            std::sync::Arc::new(std::sync::Mutex::new(job_rx));
+
+        let first_conn = &workers[0];
+        match &first_conn.conn {
+            Connection::Sql(db) => {
+                let poller_killpill_rx = rx.resubscribe();
+                let poller_name = format!(
+                    "{}-poller",
+                    first_conn
+                        .worker_name
+                        .rsplit_once('-')
+                        .map_or(first_conn.worker_name.as_str(), |(prefix, _)| prefix)
+                );
+                let db = db.clone();
+                handles.push(tokio::spawn(async move {
+                    windmill_worker::run_native_poller(
+                        &db,
+                        job_tx,
+                        semaphore,
+                        poller_killpill_rx,
+                        &poller_name,
+                    )
+                    .await;
+                }));
+                Some(native_job_rx)
+            }
+            Connection::Http(_) => {
+                tracing::warn!("native polling mode is not supported with HTTP connections, falling back to individual polling");
+                None
+            }
+        }
+    } else {
+        None
+    };
+
     for i in 1..(num_workers + 1) {
         let wk_conf = &workers[i as usize - 1];
         let conn1 = wk_conf.conn.clone();
@@ -1837,6 +1879,7 @@ pub async fn run_workers(
         let tx = tx.clone();
         let base_internal_url = base_internal_url.clone();
         let hostname = hostname.clone();
+        let native_job_rx = native_job_rx.clone();
 
         handles.push(tokio::spawn(async move {
             if num_workers > 1 {
@@ -1853,6 +1896,7 @@ pub async fn run_workers(
                 rx,
                 tx,
                 &base_internal_url,
+                native_job_rx,
             );
 
             // #[cfg(tokio_unstable)]

backend/src/monitor.rs

@@ -71,7 +71,7 @@ use windmill_common::{
     worker::{
         load_env_vars, load_init_bash_from_env, load_periodic_bash_script_from_env,
         load_periodic_bash_script_interval_from_env, load_whitelist_env_vars_from_env,
-        load_worker_config, reload_custom_tags_setting, store_pull_query,
+        load_worker_config, reload_custom_tags_setting, store_pull_query, store_pull_query_batch,
         store_suspended_pull_query, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE,
         DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR,
         WORKER_CONFIG, WORKER_GROUP,
@@ -2268,6 +2268,7 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
             tracing::info!("Reloading worker config...");
             store_suspended_pull_query(&config).await;
             store_pull_query(&config).await;
+            store_pull_query_batch(&config).await;
             *wc = config
         }
     }

backend/tests/nativets_stress.rs

@@ -230,6 +230,7 @@ fn spawn_workers(
                 }];
                 windmill_common::worker::store_suspended_pull_query(&wc).await;
                 windmill_common::worker::store_pull_query(&wc).await;
+                windmill_common::worker::store_pull_query_batch(&wc).await;
             }
             windmill_worker::run_worker(
                 &conn,
@@ -241,6 +242,7 @@ fn spawn_workers(
                 rx,
                 tx2,
                 &base_internal_url,
+                None,
             )
             .await;
         };

backend/windmill-api-configs/Cargo.toml

@@ -13,7 +13,7 @@ default = []
 enterprise = ["dep:windmill-autoscaling"]
 private = []
 python = []
-inline_preview = ["dep:windmill-worker", "dep:itertools"]
+run_inline = ["dep:windmill-worker", "dep:itertools"]
 
 [dependencies]
 windmill-api-auth.workspace = true

backend/windmill-api-configs/src/lib.rs

@@ -283,14 +283,14 @@ async fn native_kubernetes_autoscaling_healthcheck() -> Result<(), error::Error>
 }
 
 async fn list_available_python_versions() -> error::JsonResult<Vec<String>> {
-    #[cfg(not(all(feature = "python", feature = "inline_preview")))]
+    #[cfg(not(all(feature = "python", feature = "run_inline")))]
     return Err(error::Error::BadRequest(
         "Python listing available only with 'python' feature enabled".to_string(),
     ));
 
-    #[cfg(all(feature = "python", feature = "inline_preview"))]
+    #[cfg(all(feature = "python", feature = "run_inline"))]
     use itertools::Itertools;
-    #[cfg(all(feature = "python", feature = "inline_preview"))]
+    #[cfg(all(feature = "python", feature = "run_inline"))]
     return Ok(Json(
         windmill_worker::PyV::list_available_python_versions()
             .await

backend/windmill-api-integration-tests/Cargo.toml

@@ -14,6 +14,7 @@ private = ["windmill-test-utils/private", "dep:aws-config", "dep:aws-credential-
 enterprise = ["windmill-test-utils/enterprise", "dep:base64"]
 deno_core = ["windmill-test-utils/deno_core"]
 mcp = []
+run_inline = ["dep:windmill-worker", "windmill-test-utils/run_inline", "windmill-test-utils/duckdb"]
 
 [dependencies]
 windmill-test-utils.workspace = true
@@ -21,6 +22,7 @@ windmill-api-client.workspace = true
 windmill-common = { workspace = true, default-features = false }
 windmill-native-triggers = { workspace = true, features = ["native_trigger"] }
 windmill-api-auth.workspace = true
+windmill-worker = { workspace = true, optional = true }
 sqlx.workspace = true
 serde_json.workspace = true
 serde.workspace = true

backend/windmill-api-integration-tests/tests/run_inline.rs

@@ -0,0 +1,258 @@
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+
+use windmill_test_utils::*;
+
+#[cfg(feature = "run_inline")]
+async fn init_inline_utils(port: u16) -> anyhow::Result<()> {
+    use std::sync::Once;
+    static INIT: Once = Once::new();
+
+    INIT.call_once(|| {
+        let (killpill_tx, killpill_rx) = windmill_common::KillpillSender::new(1);
+        let base_internal_url = format!("http://localhost:{}", port);
+        windmill_worker::init_worker_internal_server_inline_utils(killpill_rx, base_internal_url)
+            .expect("Failed to initialize inline utils");
+        // Keep killpill_tx alive for the test duration
+        std::mem::forget(killpill_tx);
+    });
+
+    Ok(())
+}
+
+fn run_inline_url(port: u16, endpoint: &str) -> String {
+    format!("http://localhost:{port}/api/w/test-workspace/jobs/run_inline/{endpoint}")
+}
+
+fn client() -> reqwest::Client {
+    reqwest::Client::new()
+}
+
+fn authed(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
+    builder.header("Authorization", "Bearer SECRET_TOKEN")
+}
+
+fn new_script(
+    path: &str,
+    summary: &str,
+    content: &str,
+    language: &str,
+    schema_properties: serde_json::Value,
+) -> serde_json::Value {
+    json!({
+        "path": path,
+        "summary": summary,
+        "description": "",
+        "content": content,
+        "language": language,
+        "schema": {
+            "$schema": "https://json-schema.org/draft/2020-12/schema",
+            "type": "object",
+            "properties": schema_properties,
+            "required": []
+        }
+    })
+}
+
+#[cfg(feature = "run_inline")]
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_run_inline_by_path(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let base = format!("http://localhost:{port}/api/w/test-workspace");
+
+    // Initialize inline utils for script execution
+    init_inline_utils(port).await?;
+
+    // Create a DuckDB script (one of the languages that supports inline execution)
+    // DuckDB requires parameter declarations in comments: -- $param_name (type)
+    let script_path = "u/test-user/inline_test";
+    let script_content = "-- $x (integer)
+-- $y (integer)
+SELECT $x + $y as result";
+
+    let resp = authed(client().post(format!("{base}/scripts/create")))
+        .json(&new_script(
+            script_path,
+            "Inline test script",
+            script_content,
+            "duckdb",
+            json!({
+                "x": {"type": "integer"},
+                "y": {"type": "integer"}
+            }),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 201, "create script: {}", resp.text().await?);
+
+    // Test run_inline by path with args
+    let resp = authed(client().post(run_inline_url(port, &format!("p/{script_path}"))))
+        .json(&json!({
+            "args": {
+                "x": 5,
+                "y": 15
+            }
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), 200, "run_inline by path with args failed");
+    let result = resp.json::<serde_json::Value>().await?;
+    // DuckDB query should return array with one row containing result field
+    // The result structure is [{"result": 20}]
+    assert!(result.is_array(), "expected array result, got: {}", result);
+    let rows = result.as_array().unwrap();
+    assert_eq!(rows.len(), 1, "expected 1 row, got: {}", rows.len());
+    assert_eq!(
+        rows[0]["result"],
+        json!(20),
+        "expected result 20, got: {}",
+        rows[0]
+    );
+
+    Ok(())
+}
+
+#[cfg(feature = "run_inline")]
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_run_inline_by_hash(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let base = format!("http://localhost:{port}/api/w/test-workspace");
+
+    // Initialize inline utils for script execution
+    init_inline_utils(port).await?;
+
+    // Create a DuckDB script and get its hash
+    let script_path = "u/test-user/inline_hash_test";
+    let script_content = "-- $a (integer)
+-- $b (integer)
+SELECT $a * $b as product";
+
+    let resp = authed(client().post(format!("{base}/scripts/create")))
+        .json(&new_script(
+            script_path,
+            "Inline hash test script",
+            script_content,
+            "duckdb",
+            json!({
+                "a": {"type": "integer"},
+                "b": {"type": "integer"}
+            }),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 201, "create script: {}", resp.text().await?);
+
+    // Get the script to retrieve its hash
+    let resp = authed(client().get(format!("{base}/scripts/get/p/{script_path}")))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let script_data = resp.json::<serde_json::Value>().await?;
+    let hash = script_data["hash"]
+        .as_str()
+        .expect("hash should be present");
+
+    // Test run_inline by hash with args
+    let resp = authed(client().post(run_inline_url(port, &format!("h/{hash}"))))
+        .json(&json!({
+            "args": {
+                "a": 7,
+                "b": 3
+            }
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), 200, "run_inline by hash with args failed");
+    let result = resp.json::<serde_json::Value>().await?;
+    // Should return array with one row: [{"product": 21}]
+    assert!(result.is_array(), "expected array result");
+    let rows = result.as_array().unwrap();
+    assert_eq!(rows.len(), 1, "expected 1 row");
+    assert_eq!(
+        rows[0]["product"],
+        json!(21),
+        "expected product 21, got: {}",
+        rows[0]
+    );
+
+    Ok(())
+}
+
+#[cfg(feature = "run_inline")]
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_run_inline_preview(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    // Initialize inline utils for script execution
+    init_inline_utils(port).await?;
+
+    // Test run_inline preview with direct DuckDB content
+    let resp = authed(client().post(run_inline_url(port, "preview")))
+        .json(&json!({
+            "content": "-- $msg (text)\nSELECT 'Hello, ' || $msg || '!' as greeting",
+            "language": "duckdb",
+            "args": {
+                "msg": "World"
+            }
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), 200, "run_inline preview failed");
+    let result = resp.json::<serde_json::Value>().await?;
+    // Should return array with one row: [{"greeting": "Hello, World!"}]
+    assert!(result.is_array(), "expected array result");
+    let rows = result.as_array().unwrap();
+    assert_eq!(rows.len(), 1, "expected 1 row");
+    assert_eq!(
+        rows[0]["greeting"],
+        json!("Hello, World!"),
+        "expected 'Hello, World!', got: {}",
+        rows[0]
+    );
+
+    Ok(())
+}
+
+#[cfg(feature = "run_inline")]
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_run_inline_nonexistent_script(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    // Initialize inline utils
+    init_inline_utils(port).await?;
+
+    // Test run_inline by path with non-existent script - should return an error
+    let resp = authed(client().post(run_inline_url(port, "p/u/test-user/nonexistent_script")))
+        .json(&json!({
+            "args": null
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    // Should return an error (script not found)
+    assert!(
+        resp.status().is_client_error() || resp.status().is_server_error(),
+        "expected error status for nonexistent script, got: {}",
+        resp.status()
+    );
+
+    Ok(())
+}

backend/windmill-api/Cargo.toml

@@ -13,7 +13,7 @@ default = []
 private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-email?/private"]
 enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-api-scripts/enterprise", "windmill-api-flows/enterprise", "windmill-api-users/enterprise", "windmill-api-workspaces/enterprise", "windmill-api-groups/enterprise", "windmill-api-configs/enterprise", "windmill-api-settings/enterprise", "windmill-api-agent-workers?/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise"]
 stripe = []
-inline_preview = ["dep:windmill-worker", "windmill-api-configs/inline_preview"]
+run_inline = ["dep:windmill-worker", "windmill-api-configs/run_inline"]
 agent_worker_server = ["dep:windmill-worker", "dep:windmill-api-agent-workers"]
 enterprise_saml = ["dep:samael", "dep:libxml"]
 benchmark = []

backend/windmill-api/openapi.yaml

@@ -9173,6 +9173,54 @@ paths:
             application/json:
               schema: {}
 
+  /w/{workspace}/jobs/run_inline/p/{path}:
+    post:
+      summary: run script by path without starting a new job
+      operationId: runScriptByPathInline
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/ScriptPath"
+      requestBody:
+        description: script args
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/InlineScriptArgs"
+
+      responses:
+        "200":
+          description: script result
+          content:
+            application/json:
+              schema: {}
+
+  /w/{workspace}/jobs/run_inline/h/{hash}:
+    post:
+      summary: run script by hash without starting a new job
+      operationId: runScriptByHashInline
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/ScriptHash"
+      requestBody:
+        description: script args
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/InlineScriptArgs"
+
+      responses:
+        "200":
+          description: script result
+          content:
+            application/json:
+              schema: {}
+
   /w/{workspace}/jobs/run_wait_result/preview:
     post:
       summary: run script preview and wait for result
@@ -19811,6 +19859,12 @@ components:
           $ref: "#/components/schemas/ScriptLang"
       required: [content, args, language]
 
+    InlineScriptArgs:
+      type: object
+      properties:
+        args:
+          $ref: "#/components/schemas/ScriptArgs"
+
     WorkflowTask:
       type: object
       properties:

backend/windmill-api/src/jobs.rs

@@ -27,20 +27,22 @@ use url::Url;
 #[cfg(all(feature = "enterprise", feature = "smtp"))]
 use windmill_common::auth::is_super_admin_email;
 use windmill_common::auth::TOKEN_PREFIX_LEN;
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 use windmill_common::client::AuthedClient;
 use windmill_common::db::UserDbWithAuthed;
 use windmill_common::error::JsonResult;
 use windmill_common::flow_status::{JobResult, RestartedFrom};
-#[cfg(feature = "inline_preview")]
-use windmill_common::jobs::RunInlinePreviewScriptFnParams;
 use windmill_common::jobs::{
     format_completed_job_result, format_result, DynamicInput, ENTRYPOINT_OVERRIDE,
 };
+#[cfg(feature = "run_inline")]
+use windmill_common::jobs::{
+    InlineScriptTarget, RunInlinePreviewScriptFnParams, RunInlineScriptFnParams,
+};
 use windmill_common::runnable_settings::{
     ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings,
 };
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 use windmill_common::runtime_assets::{register_runtime_asset, InsertRuntimeAssetParams};
 use windmill_common::scripts::ScriptRunnableSettingsInline;
 use windmill_common::triggers::TriggerMetadata;
@@ -53,15 +55,15 @@ use windmill_common::DYNAMIC_INPUT_CACHE;
 #[cfg(all(feature = "enterprise", feature = "smtp"))]
 use windmill_common::{email_oss::send_email_html, server::load_smtp_config};
 use windmill_object_store::upload_artifact_to_store;
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 use windmill_parser::asset_parser::AssetKind;
 use windmill_types::s3::BundleFormat;
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 use windmill_worker::get_worker_internal_server_inline_utils;
 
 use windmill_common::variables::get_workspace_key;
 
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 use crate::db::OptJobAuthed;
 use crate::triggers::trigger_helpers::{FlowId, ScriptId};
 use crate::{
@@ -242,6 +244,11 @@ pub fn workspaced_service() -> Router {
         )
         .route("/run/preview", post(run_preview_script))
         .route("/run_inline/preview", post(run_inline_preview_script))
+        .route(
+            "/run_inline/p/*script_path",
+            post(run_inline_script_by_path),
+        )
+        .route("/run_inline/h/:hash", post(run_inline_script_by_hash))
         .route(
             "/run_wait_result/preview",
             post(run_wait_result_preview_script),
@@ -2853,7 +2860,7 @@ struct Preview {
     flow_path: Option<String>,
 }
 
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 #[derive(Debug, Deserialize)]
 struct PreviewInline {
     content: String,
@@ -2861,6 +2868,12 @@ struct PreviewInline {
     language: ScriptLang,
 }
 
+#[cfg(feature = "run_inline")]
+#[derive(Debug, Deserialize)]
+struct InlineScriptArgs {
+    args: Option<HashMap<String, Box<JsonRawValue>>>,
+}
+
 #[derive(Deserialize)]
 pub struct WorkflowTask {
     pub args: Option<HashMap<String, Box<JsonRawValue>>>,
@@ -4573,7 +4586,7 @@ async fn run_preview_script(
     Ok((StatusCode::CREATED, uuid.to_string()))
 }
 
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
 async fn run_inline_preview_script(
     OptJobAuthed { authed, job_id }: OptJobAuthed,
     Tokened { token }: Tokened,
@@ -4610,14 +4623,120 @@ async fn run_inline_preview_script(
     Ok(Json(to_raw_value(&result)).into_response())
 }
 
-#[cfg(not(feature = "inline_preview"))]
+#[cfg(not(feature = "run_inline"))]
 async fn run_inline_preview_script() -> error::Result<Response> {
     Err(error::Error::InternalErr(
         "inline preview requires the worker feature".to_string(),
     ))
 }
 
-#[cfg(feature = "inline_preview")]
+#[cfg(feature = "run_inline")]
+async fn run_inline_script_by_path(
+    OptJobAuthed { authed, .. }: OptJobAuthed,
+    Tokened { token }: Tokened,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, script_path)): Path<(String, StripPath)>,
+    Json(body): Json<InlineScriptArgs>,
+) -> error::Result<Response> {
+    let script_path_str = script_path.to_path();
+    check_scopes(&authed, || format!("jobs:run:scripts:{script_path_str}"))?;
+    run_inline_script_inner(
+        authed,
+        token,
+        db,
+        w_id,
+        InlineScriptTarget::Path(script_path.to_path().to_string()),
+        body.args,
+        Some(user_db),
+    )
+    .await
+}
+
+#[cfg(not(feature = "run_inline"))]
+async fn run_inline_script_by_path() -> error::Result<Response> {
+    Err(error::Error::InternalErr(
+        "inline script by path requires the worker feature".to_string(),
+    ))
+}
+
+#[cfg(feature = "run_inline")]
+async fn run_inline_script_by_hash(
+    OptJobAuthed { authed, .. }: OptJobAuthed,
+    Tokened { token }: Tokened,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, script_hash)): Path<(String, ScriptHash)>,
+    Json(body): Json<InlineScriptArgs>,
+) -> error::Result<Response> {
+    // Resolve the script path from the hash and check scopes properly
+    let hash = script_hash.0;
+    let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
+    let ScriptHashInfo { path, .. } =
+        get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash)
+            .await?
+            .prefetch_cached(&db)
+            .await?;
+
+    check_scopes(&authed, || format!("jobs:run:scripts:{path}"))?;
+
+    run_inline_script_inner(
+        authed,
+        token,
+        db,
+        w_id,
+        InlineScriptTarget::Hash(hash),
+        body.args,
+        Some(user_db),
+    )
+    .await
+}
+
+#[cfg(not(feature = "run_inline"))]
+async fn run_inline_script_by_hash() -> error::Result<Response> {
+    Err(error::Error::InternalErr(
+        "inline script by hash requires the worker feature".to_string(),
+    ))
+}
+
+#[cfg(feature = "run_inline")]
+async fn run_inline_script_inner(
+    authed: ApiAuthed,
+    token: String,
+    db: DB,
+    w_id: String,
+    target: InlineScriptTarget,
+    args: Option<HashMap<String, Box<JsonRawValue>>>,
+    user_db: Option<UserDB>,
+) -> error::Result<Response> {
+    let utils = get_worker_internal_server_inline_utils()?;
+    let authed_owned: windmill_common::db::Authed = authed.clone().into();
+    let result = utils.run_inline_script.as_ref()(RunInlineScriptFnParams {
+        target,
+        args,
+        workspace_id: w_id.clone(),
+        base_internal_url: utils.base_internal_url.clone(),
+        killpill_rx: utils.killpill_rx.resubscribe(),
+        created_by: authed.display_username().to_string(),
+        permissioned_as: username_to_permissioned_as(&authed.username),
+        permissioned_as_email: authed.email.clone(),
+        job_dir: "".to_string(),
+        worker_name: "".to_string(),
+        worker_dir: "".to_string(),
+        client: AuthedClient {
+            base_internal_url: utils.base_internal_url.clone(),
+            force_client: None,
+            token,
+            workspace: w_id,
+        },
+        conn: windmill_common::worker::Connection::Sql(db),
+        user_db: user_db.map(|udb| (udb, authed_owned)),
+    })
+    .await?;
+    Ok(Json(to_raw_value(&result)).into_response())
+}
+
+#[cfg(feature = "run_inline")]
 fn register_potential_assets_on_inline_execution(
     job_id: Uuid,
     w_id: &str,
@@ -5224,12 +5343,16 @@ async fn add_batch_jobs(
         if dedicated_worker && path.is_some() {
             windmill_common::worker::dedicated_worker_tag(&w_id, &path.clone().unwrap())
         } else {
-            format!("{}", language.as_str())
+            language
+                .tag_str(job_kind == JobKind::Dependencies)
+                .to_string()
         }
     } else if let Some(tag) = batch_info.tag {
         tag
     } else {
-        format!("{}", language.as_str())
+        language
+            .tag_str(job_kind == JobKind::Dependencies)
+            .to_string()
     };
 
     let mut tx = user_db.begin(&authed).await?;

backend/windmill-common/src/db.rs

@@ -17,6 +17,21 @@ pub struct Authed {
     pub token_prefix: Option<String>,
 }
 
+impl Authed {
+    pub fn to_authed_ref(&self) -> AuthedRef<'_> {
+        AuthedRef {
+            email: &self.email,
+            username: &self.username,
+            is_admin: &self.is_admin,
+            is_operator: &self.is_operator,
+            groups: &self.groups,
+            folders: &self.folders,
+            scopes: &self.scopes,
+            token_prefix: &self.token_prefix,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Hash)]
 pub struct AuthedRef<'a> {
     pub email: &'a str,

backend/windmill-common/src/jobs.rs

@@ -326,6 +326,28 @@ pub struct RunInlinePreviewScriptFnParams {
     pub killpill_rx: tokio::sync::broadcast::Receiver<()>,
 }
 
+pub enum InlineScriptTarget {
+    Path(String),
+    Hash(i64),
+}
+
+pub struct RunInlineScriptFnParams {
+    pub workspace_id: String,
+    pub target: InlineScriptTarget,
+    pub args: Option<HashMap<String, Box<RawValue>>>,
+    pub created_by: String,
+    pub permissioned_as: String,
+    pub permissioned_as_email: String,
+    pub base_internal_url: String,
+    pub worker_name: String,
+    pub conn: crate::worker::Connection,
+    pub client: AuthedClient,
+    pub job_dir: String,
+    pub worker_dir: String,
+    pub killpill_rx: tokio::sync::broadcast::Receiver<()>,
+    pub user_db: Option<(crate::db::UserDB, crate::db::Authed)>,
+}
+
 #[derive(Clone)]
 pub struct WorkerInternalServerInlineUtils {
     pub killpill_rx: Arc<tokio::sync::broadcast::Receiver<()>>,
@@ -337,6 +359,13 @@ pub struct WorkerInternalServerInlineUtils {
             + Send
             + Sync,
     >,
+    pub run_inline_script: Arc<
+        dyn Fn(
+                RunInlineScriptFnParams,
+            ) -> Pin<Box<dyn Future<Output = error::Result<Box<RawValue>>> + Send>>
+            + Send
+            + Sync,
+    >,
 }
 // To run a script inline, bypassing the db and job queue, windmill-api uses these functions.
 // They should only be called by the internal server of a worker.

backend/windmill-common/src/worker.rs

@@ -233,6 +233,7 @@ lazy_static::lazy_static! {
     }));
 
     pub static ref WORKER_PULL_QUERIES: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
+    pub static ref WORKER_PULL_QUERIES_BATCH: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
     pub static ref WORKER_SUSPENDED_PULL_QUERY: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
 
 
@@ -490,6 +491,69 @@ pub async fn store_pull_query(wc: &WorkerConfig) {
     *l = queries;
 }
 
+fn format_pull_query_batch(peek: String) -> String {
+    format!(
+        "WITH peek AS (
+            {}
+        ), q AS NOT MATERIALIZED (
+            UPDATE v2_job_queue SET
+                running = true,
+                started_at = coalesce(started_at, now()),
+                suspend_until = null,
+                worker = $1
+            WHERE id = ANY(ARRAY(SELECT id FROM peek))
+            RETURNING
+                id, started_at, scheduled_for,
+                canceled_by, canceled_reason, worker, cache_ignore_s3_path, runnable_settings_handle
+        ), r AS NOT MATERIALIZED (
+            UPDATE v2_job_runtime SET
+                ping = now()
+            WHERE id = ANY(ARRAY(SELECT id FROM q))
+        ) SELECT q.id, j.workspace_id, j.parent_job, j.created_by, q.started_at, q.scheduled_for,
+            j.runnable_id, j.runnable_path, j.args, q.canceled_by,
+            q.canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as,
+            flow_status, j.script_lang,
+            j.same_worker, j.pre_run_error, j.visible_to_owner,
+            j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job,
+            j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
+            j.script_entrypoint_override, j.preprocessed, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path,
+            COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
+            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
+        FROM q JOIN v2_job j ON j.id = q.id
+            LEFT JOIN v2_job_status f ON f.id = q.id
+            LEFT JOIN job_perms p ON p.job_id = q.id
+            LEFT JOIN v2_job pj ON j.parent_job = pj.id
+            ",
+        peek
+    )
+}
+
+pub fn make_pull_query_batch(tags: &[String]) -> String {
+    format_pull_query_batch(format!(
+        "SELECT id
+        FROM v2_job_queue
+        WHERE running = false AND tag IN ({}) AND scheduled_for <= now()
+        ORDER BY priority DESC NULLS LAST, scheduled_for
+        FOR UPDATE SKIP LOCKED
+        LIMIT $2",
+        tags.iter().map(|x| format!("'{x}'")).join(", ")
+    ))
+}
+
+pub async fn store_pull_query_batch(wc: &WorkerConfig) {
+    let mut queries = vec![];
+    for tags in wc.priority_tags_sorted.iter() {
+        if tags.tags.is_empty() {
+            tracing::error!("Empty tags in priority tags, skipping");
+            continue;
+        }
+        let query = make_pull_query_batch(&tags.tags);
+        queries.push(query);
+    }
+    let mut l = WORKER_PULL_QUERIES_BATCH.write().await;
+    *l = queries;
+}
+
 pub const TMP_DIR: &str = "/tmp/windmill";
 pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs");
 

backend/windmill-queue/src/jobs.rs

@@ -74,7 +74,7 @@ use windmill_common::{
     utils::{not_found_if_none, report_critical_error, StripPath, WarnAfterExt},
     worker::{
         to_raw_value, CLOUD_HOSTED, DISABLE_FLOW_SCRIPT, NO_LOGS, WORKER_PULL_QUERIES,
-        WORKER_SUSPENDED_PULL_QUERY,
+        WORKER_PULL_QUERIES_BATCH, WORKER_SUSPENDED_PULL_QUERY,
     },
     DB, METRICS_ENABLED,
 };
@@ -2285,6 +2285,59 @@ pub struct MiniPulledJob {
     pub runnable_settings_handle: Option<i64>,
 }
 
+impl MiniPulledJob {
+    pub fn new_inline(
+        workspace_id: String,
+        args: Option<HashMap<String, Box<RawValue>>>,
+        created_by: String,
+        permissioned_as: String,
+        permissioned_as_email: String,
+        runnable_path: Option<String>,
+        kind: JobKind,
+        runnable_id: Option<ScriptHash>,
+        tag: String,
+        script_lang: Option<ScriptLang>,
+    ) -> Self {
+        Self {
+            workspace_id,
+            id: Uuid::new_v4(),
+            args: args.map(Json),
+            parent_job: None,
+            created_by,
+            scheduled_for: chrono::Utc::now(),
+            started_at: None,
+            runnable_path,
+            kind,
+            runnable_id,
+            canceled_reason: None,
+            canceled_by: None,
+            permissioned_as,
+            permissioned_as_email,
+            flow_status: None,
+            tag,
+            script_lang,
+            same_worker: true,
+            pre_run_error: None,
+            flow_innermost_root_job: None,
+            root_job: None,
+            timeout: None,
+            flow_step_id: None,
+            cache_ttl: None,
+            cache_ignore_s3_path: None,
+            priority: None,
+            preprocessed: None,
+            script_entrypoint_override: None,
+            trigger: None,
+            trigger_kind: None,
+            visible_to_owner: false,
+            permissioned_as_end_user_email: None,
+            runnable_settings_handle: None,
+            concurrent_limit: None,
+            concurrency_time_window_s: None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct MiniCompletedJob {
     pub id: Uuid,
@@ -3381,6 +3434,65 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>(
     Ok(job_and_suspended)
 }
 
+pub async fn pull_batch(
+    db: &Pool<Postgres>,
+    worker_name: &str,
+    limit: i32,
+) -> windmill_common::error::Result<Vec<PulledJob>> {
+    let queries = WORKER_PULL_QUERIES_BATCH.read().await;
+
+    if queries.is_empty() {
+        tracing::warn!("No batch pull queries available");
+        return Ok(vec![]);
+    }
+
+    let mut results: Vec<PulledJob> = Vec::new();
+    let mut remaining = limit;
+
+    for query in queries.iter() {
+        if remaining <= 0 {
+            break;
+        }
+
+        let jobs = timeout(
+            Duration::from_secs(15),
+            sqlx::query_as::<_, PulledJob>(query)
+                .bind(worker_name)
+                .bind(remaining)
+                .fetch_all(db),
+        )
+        .await??;
+
+        for job in jobs {
+            if job.is_flow() || job.is_dependency() {
+                let per_workspace = per_workspace_tag(&job.workspace_id).await;
+                let base_tag = if job.is_flow() {
+                    "flow".to_string()
+                } else {
+                    "dependency".to_string()
+                };
+                let tag = if per_workspace {
+                    format!("{}-{}", base_tag, job.workspace_id)
+                } else {
+                    base_tag
+                };
+                sqlx::query!(
+                    "UPDATE v2_job_queue SET tag = $1, running = false WHERE id = $2",
+                    tag,
+                    job.id
+                )
+                .execute(db)
+                .await?;
+                continue;
+            }
+            remaining -= 1;
+            results.push(job);
+        }
+    }
+
+    Ok(results)
+}
+
 pub async fn custom_concurrency_key(
     db: &Pool<Postgres>,
     job_id: &Uuid,
@@ -5320,15 +5432,7 @@ async fn push_inner<'c, 'd>(
             language
                 .as_ref()
                 .map(|x| {
-                    let tag_lang = if x == &ScriptLang::Bunnative {
-                        if job_kind == JobKind::Dependencies {
-                            ScriptLang::Bun.as_str()
-                        } else {
-                            ScriptLang::Nativets.as_str()
-                        }
-                    } else {
-                        x.as_str()
-                    };
+                    let tag_lang = x.tag_str(job_kind == JobKind::Dependencies);
                     if per_workspace {
                         format!("{}-{}", tag_lang, workspace_id)
                     } else {

backend/windmill-test-utils/Cargo.toml

@@ -15,6 +15,8 @@ enterprise = []
 python = ["windmill-common/python"]
 deno_core = ["dep:windmill-runtime-nativets"]
 agent_worker_server = ["dep:windmill-api-agent-workers"]
+run_inline = ["windmill-api/run_inline"]
+duckdb = ["windmill-worker/duckdb"]
 
 [dependencies]
 windmill-api = { workspace = true, default-features = false }

backend/windmill-test-utils/src/lib.rs

@@ -401,6 +401,7 @@ pub fn spawn_test_worker(
             }];
             windmill_common::worker::store_suspended_pull_query(&wc).await;
             windmill_common::worker::store_pull_query(&wc).await;
+            windmill_common::worker::store_pull_query_batch(&wc).await;
         }
         windmill_worker::run_worker(
             &conn,
@@ -412,6 +413,7 @@ pub fn spawn_test_worker(
             rx,
             tx2,
             &base_internal_url,
+            None,
         )
         .await
     };

backend/windmill-types/src/scripts.rs

@@ -88,6 +88,18 @@ impl ScriptLang {
         }
     }
 
+    pub fn tag_str(&self, is_dependency: bool) -> &'static str {
+        if self == &ScriptLang::Bunnative {
+            if is_dependency {
+                ScriptLang::Bun.as_str()
+            } else {
+                ScriptLang::Nativets.as_str()
+            }
+        } else {
+            self.as_str()
+        }
+    }
+
     pub fn as_dependencies_filename(&self) -> Option<String> {
         use ScriptLang::*;
         Some(
@@ -105,15 +117,15 @@ impl ScriptLang {
     pub fn is_native(&self) -> bool {
         matches!(
             self,
-            ScriptLang::Bunnative |
-            ScriptLang::Nativets |
-            ScriptLang::Postgresql |
-            ScriptLang::Mysql |
-            ScriptLang::Graphql |
-            ScriptLang::Snowflake |
-            ScriptLang::Mssql |
-            ScriptLang::Bigquery |
-            ScriptLang::OracleDB
+            ScriptLang::Bunnative
+                | ScriptLang::Nativets
+                | ScriptLang::Postgresql
+                | ScriptLang::Mysql
+                | ScriptLang::Graphql
+                | ScriptLang::Snowflake
+                | ScriptLang::Mssql
+                | ScriptLang::Bigquery
+                | ScriptLang::OracleDB
         )
     }
 

backend/windmill-worker/src/worker.rs

@@ -14,6 +14,10 @@ use futures::TryFutureExt;
 use tokio::sync::Mutex;
 use tokio::time::timeout;
 use windmill_common::client::AuthedClient;
+use windmill_common::db::UserDbWithAuthed;
+use windmill_common::get_latest_deployed_hash_for_path;
+use windmill_common::jobs::InlineScriptTarget;
+use windmill_common::jobs::RunInlineScriptFnParams;
 use windmill_common::jobs::WorkerInternalServerInlineUtils;
 use windmill_common::jobs::WORKER_INTERNAL_SERVER_INLINE_UTILS;
 use windmill_common::runtime_assets::init_runtime_asset_loop;
@@ -68,6 +72,7 @@ use windmill_parser::MainArgSignature;
 use windmill_queue::DedicatedWorkerJob;
 use windmill_queue::FlowRunners;
 use windmill_queue::MiniCompletedJob;
+use windmill_queue::PulledJobResult;
 use windmill_queue::PulledJobResultToJobErr;
 
 use uuid::Uuid;
@@ -85,8 +90,8 @@ use windmill_common::{
 };
 
 use windmill_queue::{
-    append_logs, canceled_job_to_result, empty_result, get_same_worker_job, pull, push_init_job,
-    push_periodic_bash_job, CanceledBy, JobAndPerms, JobCompleted, MiniPulledJob,
+    append_logs, canceled_job_to_result, empty_result, get_same_worker_job, pull, pull_batch,
+    push_init_job, push_periodic_bash_job, CanceledBy, JobAndPerms, JobCompleted, MiniPulledJob,
     PrecomputedAgentInfo, PulledJob, SameWorkerPayload, HTTP_CLIENT, INIT_SCRIPT_TAG,
     PERIODIC_SCRIPT_TAG,
 };
@@ -252,6 +257,13 @@ const VACUUM_PERIOD: u32 = 10000;
 
 pub const MAX_BUFFERED_DEDICATED_JOBS: usize = 3;
 
+pub struct NativeJobDispatch {
+    pub job: PulledJobResult,
+    pub permit: tokio::sync::OwnedSemaphorePermit,
+}
+
+pub type NativeJobRx = Arc<std::sync::Mutex<tokio::sync::mpsc::Receiver<NativeJobDispatch>>>;
+
 /// Per-language OTEL tracing proxy configuration.
 /// Default languages are configured in frontend instanceSettings.ts
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -1344,6 +1356,87 @@ pub async fn create_job_dir(worker_directory: &str, job_id: impl Display) -> Str
     job_dir_path
 }
 
+pub async fn run_native_poller(
+    db: &sqlx::Pool<sqlx::Postgres>,
+    job_tx: tokio::sync::mpsc::Sender<NativeJobDispatch>,
+    semaphore: Arc<tokio::sync::Semaphore>,
+    mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
+    worker_name: &str,
+) {
+    loop {
+        match killpill_rx.try_recv() {
+            Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Closed) => {
+                tracing::info!(worker = %worker_name, "native poller: killpill received, exiting");
+                break;
+            }
+            _ => {}
+        }
+
+        let available = semaphore.available_permits();
+        if available == 0 {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+            continue;
+        }
+
+        let jobs = match timeout(
+            Duration::from_secs(30),
+            pull_batch(db, worker_name, available as i32),
+        )
+        .await
+        {
+            Ok(Ok(jobs)) => jobs,
+            Ok(Err(e)) => {
+                tracing::error!(worker = %worker_name, "native poller: pull_batch error: {e}");
+                tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await;
+                continue;
+            }
+            Err(_) => {
+                tracing::error!(worker = %worker_name, "native poller: pull_batch timed out");
+                continue;
+            }
+        };
+
+        if jobs.is_empty() {
+            tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await;
+            continue;
+        }
+
+        for pulled_job in jobs {
+            let permit = semaphore
+                .clone()
+                .try_acquire_owned()
+                .expect("permits can only increase between available_permits() and here");
+
+            let mut job = PulledJobResult {
+                job: Some(pulled_job),
+                suspended: false,
+                missing_concurrency_key: false,
+                error_while_preprocessing: None,
+            };
+
+            if let Err(e) = timeout(
+                core::time::Duration::from_secs(10),
+                job.maybe_apply_debouncing(db),
+            )
+            .await
+            .map_err(error::Error::from)
+            .and_then(|r| r)
+            {
+                job.error_while_preprocessing = Some(e.to_string());
+            }
+
+            if job_tx
+                .send(NativeJobDispatch { job, permit })
+                .await
+                .is_err()
+            {
+                tracing::info!(worker = %worker_name, "native poller: channel closed, exiting");
+                return;
+            }
+        }
+    }
+}
+
 pub async fn run_worker(
     conn: &Connection,
     hostname: &str,
@@ -1354,6 +1447,7 @@ pub async fn run_worker(
     mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
     killpill_tx: KillpillSender,
     base_internal_url: &str,
+    native_job_rx: Option<NativeJobRx>,
 ) {
     #[cfg(not(feature = "enterprise"))]
     if is_sandboxing_enabled() {
@@ -1814,10 +1908,13 @@ pub async fn run_worker(
     let mut last_30jobs_suspended = 0;
     let mut last_suspend_first = Instant::now();
     let mut killed_but_draining_same_worker_jobs = false;
+    let mut native_permit: Option<tokio::sync::OwnedSemaphorePermit> = None;
 
     let mut killpill_rx2 = killpill_rx.resubscribe();
 
     loop {
+        drop(native_permit.take());
+
         let last_processing_duration_secs = last_processing_duration.load(Ordering::SeqCst);
         if last_processing_duration_secs > 5 {
             let sleep_duration = if last_processing_duration_secs > 10 {
@@ -2053,6 +2150,27 @@ pub async fn run_worker(
                     tokio::time::sleep(Duration::from_millis(200)).await;
                     continue;
                 }
+            } else if let Some(ref native_rx) = native_job_rx {
+                let dispatch = native_rx.lock().unwrap().try_recv().ok();
+                match dispatch {
+                    Some(d) => {
+                        native_permit = Some(d.permit);
+                        match d.job.to_pulled_job() {
+                            Ok(j) => Ok(j.map(|job| NextJob::Sql { flow_runners: None, job })),
+                            Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc))
+                            | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(jc)) => {
+                                if let Err(err) = job_completed_tx.send_job(jc, true).await {
+                                    tracing::error!(
+                                        "An error occurred while sending job completed: {:#?}",
+                                        err
+                                    )
+                                }
+                                Ok(None)
+                            }
+                        }
+                    }
+                    None => Ok(None),
+                }
             } else {
                 match &conn {
                     Connection::Sql(db) => {
@@ -4705,43 +4823,18 @@ pub fn init_worker_internal_server_inline_utils(
         base_internal_url,
         killpill_rx: Arc::new(killpill_rx),
         run_inline_preview_script: Arc::new(|params| {
-            let job = MiniPulledJob {
-                workspace_id: params.workspace_id,
-                id: Uuid::new_v4(),
-                args: params.args.map(Json),
-                parent_job: None,
-                created_by: params.created_by,
-                scheduled_for: chrono::Utc::now(),
-                started_at: None,
-                runnable_path: None,
-                kind: JobKind::Preview,
-                runnable_id: None,
-                canceled_reason: None,
-                canceled_by: None,
-                permissioned_as: params.permissioned_as,
-                permissioned_as_email: params.permissioned_as_email,
-                flow_status: None,
-                tag: "inline_preview".to_string(),
-                script_lang: Some(params.lang),
-                same_worker: true,
-                pre_run_error: None,
-                flow_innermost_root_job: None,
-                root_job: None,
-                timeout: None,
-                flow_step_id: None,
-                cache_ttl: None,
-                cache_ignore_s3_path: None,
-                priority: None,
-                preprocessed: None,
-                script_entrypoint_override: None,
-                trigger: None,
-                trigger_kind: None,
-                visible_to_owner: false,
-                permissioned_as_end_user_email: None,
-                runnable_settings_handle: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
-            };
+            let job = MiniPulledJob::new_inline(
+                params.workspace_id,
+                params.args,
+                params.created_by,
+                params.permissioned_as,
+                params.permissioned_as_email,
+                None,
+                JobKind::Preview,
+                None,
+                "inline_preview".to_string(),
+                Some(params.lang),
+            );
             Box::pin(async move {
                 let mut mem_peak: i32 = -1;
                 let mut canceled_by: Option<CanceledBy> = None;
@@ -4778,6 +4871,86 @@ pub fn init_worker_internal_server_inline_utils(
                 .await
             })
         }),
+        run_inline_script: Arc::new(|params: RunInlineScriptFnParams| {
+            Box::pin(async move {
+                let (script_hash, runnable_path) = match params.target {
+                    InlineScriptTarget::Path(ref path) => {
+                        let db = params
+                            .conn
+                            .as_sql()
+                            .ok_or_else(|| {
+                                error::Error::InternalErr(
+                                    "run_inline_script by path requires a SQL connection"
+                                        .to_string(),
+                                )
+                            })?
+                            .clone();
+                        let authed_ref = params.user_db.as_ref().map(|(_, a)| a.to_authed_ref());
+                        let user_db_authed =
+                            params.user_db.as_ref().zip(authed_ref.as_ref()).map(
+                                |((udb, _), ar)| UserDbWithAuthed { db: udb.clone(), authed: ar },
+                            );
+                        let script_hash_info = get_latest_deployed_hash_for_path(
+                            user_db_authed,
+                            db,
+                            &params.workspace_id,
+                            path,
+                        )
+                        .await?;
+                        (ScriptHash(script_hash_info.hash), Some(path.clone()))
+                    }
+                    InlineScriptTarget::Hash(hash) => (ScriptHash(hash), None),
+                };
+                let content_info =
+                    get_script_content_by_hash(&script_hash, &params.workspace_id, &params.conn)
+                        .await?;
+                let job = MiniPulledJob::new_inline(
+                    params.workspace_id,
+                    params.args,
+                    params.created_by,
+                    params.permissioned_as,
+                    params.permissioned_as_email,
+                    runnable_path,
+                    JobKind::Script,
+                    Some(script_hash),
+                    "inline_run".to_string(),
+                    content_info.language,
+                );
+                let mut mem_peak: i32 = -1;
+                let mut canceled_by: Option<CanceledBy> = None;
+                let mut column_order: Option<Vec<String>> = None;
+                let mut new_args: Option<HashMap<String, Box<RawValue>>> = None;
+                let mut occupancy_metrics = OccupancyMetrics::new(Instant::now());
+                let mut has_stream: bool = false;
+                let mut killpill_rx = params.killpill_rx;
+
+                run_language_executor(
+                    &job,
+                    &params.conn,
+                    &params.client,
+                    None,
+                    &params.job_dir,
+                    &params.worker_dir,
+                    &mut mem_peak,
+                    &mut canceled_by,
+                    &params.base_internal_url,
+                    &params.worker_name,
+                    &mut column_order,
+                    &mut new_args,
+                    &mut occupancy_metrics,
+                    &mut killpill_rx,
+                    None,
+                    &mut has_stream,
+                    content_info.language,
+                    &content_info.content,
+                    &content_info.envs,
+                    &content_info.codebase,
+                    &content_info.lockfile,
+                    true,
+                )
+                .await
+            })
+        }),
     };
     WORKER_INTERNAL_SERVER_INLINE_UTILS
         .set(utils)

backend/write_latest_ee_ref.sh

@@ -1,7 +1,19 @@
 #!/bin/bash
 
-# Navigate to the target directory
-cd ../../windmill-ee-private || cd ~/windmill-ee-private || { echo "Directory not found"; exit 1; }
+# Detect the current branch name
+branch=$(git rev-parse --abbrev-ref HEAD 2>/dev/null)
+
+# Try, in order: matching EE worktree, sibling repo, home directory fallback
+ee_worktree="$HOME/windmill-ee-private__worktrees/$branch"
+if [ -n "$branch" ] && [ -d "$ee_worktree" ]; then
+    cd "$ee_worktree"
+elif cd ../../windmill-ee-private 2>/dev/null; then
+    :
+elif cd ~/windmill-ee-private 2>/dev/null; then
+    :
+else
+    echo "Directory not found"; exit 1
+fi
 
 # Get the current commit hash
 commit_hash=$(git rev-parse HEAD)

frontend/src/lib/components/RunsPage.svelte

@@ -87,6 +87,8 @@
 		})
 	)
 	let perPage = useLocalStorageValue('runs_per_page', 1000, 'number')
+	let showSchedulesStorage = useLocalStorageValue('runs_show_schedules', true, 'boolean')
+	let showFutureJobsStorage = useLocalStorageValue('runs_show_future_jobs', true, 'boolean')
 	let filters = useUrlSyncedFilterInstance(untrack(() => runsFilterSearchbarSchema))
 
 	let { initialPath }: Props = $props()
@@ -98,6 +100,24 @@
 		filters.val.path = initialPath
 	}
 
+	// Apply persistent toggle values from local storage if URL doesn't specify them
+	if (!page.url.searchParams.has('job_trigger_kind') && showSchedulesStorage.val === false) {
+		filters.val.job_trigger_kind = '!schedule'
+	}
+	if (!page.url.searchParams.has('show_future_jobs') && showFutureJobsStorage.val === false) {
+		filters.val.show_future_jobs = false
+	}
+
+	// Sync toggle state back to local storage when filters change
+	$effect(() => {
+		if (!filters.val.job_trigger_kind || filters.val.job_trigger_kind === '!schedule') {
+			showSchedulesStorage.val = filters.val.job_trigger_kind !== '!schedule'
+		}
+	})
+	$effect(() => {
+		showFutureJobsStorage.val = filters.val.show_future_jobs !== false
+	})
+
 	let selectedIds: string[] = $state([])
 	let selectedWorkspace: string | undefined = $state(undefined)
 

scripts/wm-cursor

@@ -10,25 +10,30 @@ setopt err_exit no_unset pipe_fail
 
 local script_path=${0:A}
 
-# --- Resolve Cursor remote CLI (most recently modified) ---
+# --- Lazy Cursor CLI resolution (only when needed) ---
 
-local -a cursor_bins=(~/.cursor-server/cli/servers/*/server/bin/remote-cli/cursor(NOm))
-if (( ${#cursor_bins} == 0 )); then
-  print -u2 "Error: Cursor remote CLI not found in ~/.cursor-server/cli/servers/"
-  exit 1
-fi
-local cursor_bin=${cursor_bins[1]}
+local cursor_bin=
 
-# --- Refresh Cursor IPC socket (tmux may hold a stale one) ---
-# Multiple stale sockets may exist; probe to find a live one
+resolve_cursor_cli() {
+  [[ -n $cursor_bin ]] && return 0
 
-local sock
-for sock in /tmp/vscode-ipc-*.sock(NOm); do
-  if VSCODE_IPC_HOOK_CLI=$sock $cursor_bin --status &>/dev/null; then
-    export VSCODE_IPC_HOOK_CLI=$sock
-    break
+  local -a cursor_bins=(~/.cursor-server/cli/servers/*/server/bin/remote-cli/cursor(NOm))
+  if (( ${#cursor_bins} == 0 )); then
+    print -u2 "Error: Cursor remote CLI not found in ~/.cursor-server/cli/servers/"
+    exit 1
   fi
-done
+  cursor_bin=${cursor_bins[1]}
+
+  # Refresh Cursor IPC socket (tmux may hold a stale one)
+  # Multiple stale sockets may exist; probe to find a live one
+  local sock
+  for sock in /tmp/vscode-ipc-*.sock(NOm); do
+    if timeout 2 env VSCODE_IPC_HOOK_CLI=$sock $cursor_bin --status &>/dev/null; then
+      export VSCODE_IPC_HOOK_CLI=$sock
+      break
+    fi
+  done
+}
 
 # --- Helper functions ---
 
@@ -133,6 +138,7 @@ parse_features_flag() {
 # --- Subcommands ---
 
 cmd_add() {
+  resolve_cursor_cli
   ensure_tmux
   check_dev_db
 
@@ -166,6 +172,7 @@ cmd_open() {
   local name=${1:?Usage: wm-cursor open <name>}
   shift
 
+  resolve_cursor_cli
   ensure_tmux
   check_dev_db
 
@@ -202,6 +209,7 @@ cmd_close() {
 cmd_open_ee() {
   local name=${1:?Usage: wm-cursor open-ee <name>}
 
+  resolve_cursor_cli
   local wt_path=$(workmux path $name)
   local main_repo_root="$(cd "$(git -C "$wt_path" rev-parse --git-common-dir 2>/dev/null)/.." && pwd)"
 

typescript-client/client.ts

@@ -1195,6 +1195,8 @@ interface TeamsApprovalOptions {
  * @param {string} [options.approver] - Optional user ID or name of the approver for the request.
  * @param {DefaultArgs} [options.defaultArgsJson] - Optional object defining or overriding the default arguments to a form field.
  * @param {Enums} [options.dynamicEnumsJson] - Optional object overriding the enum default values of an enum form field.
+ * @param {string} [options.resumeButtonText] - Optional text for the resume button.
+ * @param {string} [options.cancelButtonText] - Optional text for the cancel button.
  *
  * @returns {Promise<void>} Resolves when the Slack approval request is successfully sent.
  *