perf: add agent-batch mode with batch pull from server

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/src/main.rs

@@ -754,7 +754,7 @@ async fn windmill_main() -> anyhow::Result<()> {
         .and_then(|x| x.parse().ok())
         .unwrap_or(IpAddr::from(default_bind_addr));
 
-    let (conn, first_suffix, agent_config) = if mode == Mode::Agent {
+    let (conn, first_suffix, agent_config) = if matches!(mode, Mode::Agent | Mode::AgentBatch) {
         let agent_config = match AgentConfig::from_env() {
             Ok(config) => config,
             Err(e) => {
@@ -829,7 +829,7 @@ async fn windmill_main() -> anyhow::Result<()> {
 
     let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode, &environment);
 
-    let is_agent = mode == Mode::Agent;
+    let is_agent = matches!(mode, Mode::Agent | Mode::AgentBatch);
 
     let mut migration_handle: Option<JoinHandle<()>> = None;
     #[cfg(feature = "parquet")]
@@ -953,7 +953,7 @@ async fn windmill_main() -> anyhow::Result<()> {
         }
     }
 
-    let conn = if mode == Mode::Agent {
+    let conn = if matches!(mode, Mode::Agent | Mode::AgentBatch) {
         conn
     } else {
         // Drop the initial connection pool before creating the main one.
@@ -1289,7 +1289,7 @@ Windmill Community Edition {GIT_VERSION}
                                 )
                             },
                             worker_name: worker_name_with_suffix(
-                                mode == Mode::Agent,
+                                matches!(mode, Mode::Agent | Mode::AgentBatch),
                                 WORKER_GROUP.as_str(),
                                 &suffix,
                             ),

backend/windmill-common/src/utils.rs

@@ -122,6 +122,21 @@ lazy_static::lazy_static! {
                 }
                 #[cfg(feature = "enterprise")]
                 Mode::Agent
+            } else if &x == "agent-batch" {
+                println!("Binary is in 'agent-batch' mode with BASE_INTERNAL_URL={}", std::env::var("BASE_INTERNAL_URL").unwrap_or_default());
+                if std::env::var("BASE_INTERNAL_URL").is_err() {
+                    panic!("BASE_INTERNAL_URL is required in agent-batch mode")
+                }
+                if std::env::var("AGENT_TOKEN").is_err() {
+                    println!("AGENT_TOKEN is not passed. This is required for the agent to work and contains the JWT to authenticate with the server.")
+                }
+
+                #[cfg(not(feature = "enterprise"))]
+                {
+                    panic!("Agent-batch mode is only available in the EE, ignoring...");
+                }
+                #[cfg(feature = "enterprise")]
+                Mode::AgentBatch
             } else if &x == "indexer" {
                 tracing::info!("Binary is in 'indexer' mode");
                 #[cfg(not(feature = "tantivy"))]
@@ -474,6 +489,7 @@ pub fn map_string_to_number(s: &str, max_number: u64) -> u64 {
 pub enum Mode {
     Worker,
     Agent,
+    AgentBatch,
     Server,
     Standalone,
     Indexer,
@@ -485,6 +501,7 @@ impl std::fmt::Display for Mode {
         match self {
             Mode::Worker => write!(f, "worker"),
             Mode::Agent => write!(f, "agent"),
+            Mode::AgentBatch => write!(f, "agent-batch"),
             Mode::Server => write!(f, "server"),
             Mode::Standalone => write!(f, "standalone"),
             Mode::Indexer => write!(f, "indexer"),

backend/windmill-worker/src/agent_workers.rs

@@ -37,6 +37,15 @@ pub async fn pull_job(
         .await
 }
 
+pub async fn batch_pull_jobs(
+    client: &HttpClient,
+    batch_size: i32,
+) -> anyhow::Result<Vec<JobAndPerms>> {
+    client
+        .post("/api/agent_workers/batch_pull", None, &batch_size)
+        .await
+}
+
 pub async fn send_result(client: &HttpClient, jc: JobCompleted) -> anyhow::Result<String> {
     client
         .post(

backend/windmill-worker/src/worker.rs

@@ -2078,6 +2078,8 @@ pub async fn run_worker(
     let batch_pull_size = *windmill_common::worker::BATCH_PULL_SIZE;
     let mut batch_pull_buffer: std::collections::VecDeque<windmill_queue::PulledJob> =
         std::collections::VecDeque::new();
+    let mut agent_batch_buffer: std::collections::VecDeque<windmill_queue::JobAndPerms> =
+        std::collections::VecDeque::new();
 
     let mut killpill_rx2 = killpill_rx.resubscribe();
 
@@ -2504,10 +2506,36 @@ pub async fn run_worker(
                         }
                     }
 
-                    Connection::Http(client) => crate::agent_workers::pull_job(&client, None, None)
-                        .await
-                        .map_err(|e| error::Error::InternalErr(e.to_string()))
-                        .map(|x| x.map(|y| NextJob::Http(y))),
+                    Connection::Http(client) => {
+                        if batch_pull_size > 0 {
+                            // Agent-batch mode: serve from buffer, refill via batch pull
+                            if let Some(job) = agent_batch_buffer.pop_front() {
+                                Ok(Some(NextJob::Http(job)))
+                            } else {
+                                match crate::agent_workers::batch_pull_jobs(
+                                    &client,
+                                    batch_pull_size,
+                                )
+                                .await
+                                {
+                                    Ok(mut jobs) if !jobs.is_empty() => {
+                                        let first = jobs.remove(0);
+                                        for job in jobs {
+                                            agent_batch_buffer.push_back(job);
+                                        }
+                                        Ok(Some(NextJob::Http(first)))
+                                    }
+                                    Ok(_) => Ok(None),
+                                    Err(e) => Err(error::Error::InternalErr(e.to_string())),
+                                }
+                            }
+                        } else {
+                            crate::agent_workers::pull_job(&client, None, None)
+                                .await
+                                .map_err(|e| error::Error::InternalErr(e.to_string()))
+                                .map(|x| x.map(|y| NextJob::Http(y)))
+                        }
+                    }
                 }
             }
         };