fix: treat concurrent_limit/timeout <= 0 as unset instead of a zero cap (#10288)

* fix: treat concurrent_limit/timeout <= 0 as unset instead of a zero cap

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

* fix: flow-step timeout <= 0 inherits the script timeout, not the global default

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/windmill-api-flows/src/flows.rs

@@ -527,6 +527,11 @@ async fn create_flow(
     }
     check_scopes(&authed, || format!("flows:write:{}", nf.path))?;
 
+    // A `<= 0` flow timeout is "unset", not a 0-second limit that kills every run instantly.
+    // (The concurrency settings inside the flow value are normalized on deserialization; see
+    // ConcurrencySettings.) Runtime guards also protect already-stored rows.
+    nf.timeout = windmill_common::runnable_settings::none_if_non_positive(nf.timeout);
+
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
         AuditAuthorable::username(&authed),
@@ -1012,7 +1017,9 @@ async fn update_flow(
     }
     let flow_path = flow_path.to_path();
     // The URL identifies the flow being updated; the body path is only needed to rename.
-    let nf = ef.into_new_flow(flow_path);
+    let mut nf = ef.into_new_flow(flow_path);
+    // A `<= 0` flow timeout is "unset", not a 0-second limit (see create_flow).
+    nf.timeout = windmill_common::runnable_settings::none_if_non_positive(nf.timeout);
     check_scopes(&authed, || format!("flows:write:{}", flow_path))?;
 
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(

backend/windmill-api-scripts/src/scripts.rs

@@ -914,6 +914,13 @@ async fn create_script_internal<'c>(
     }
     check_scopes(&authed, || format!("scripts:write:{}", ns.path))?;
 
+    // Normalize positive-only settings so a `<= 0` value (e.g. a CLI-pushed `0`) persists as
+    // disabled rather than as a zero-slot concurrency cap or a 0-second timeout. Deserialization
+    // already normalizes the concurrency fields; re-applying here also covers `timeout` and any
+    // NewScript built in-process rather than from a request body.
+    ns.timeout = windmill_common::runnable_settings::none_if_non_positive(ns.timeout);
+    ns.concurrency_settings = ns.concurrency_settings.normalized();
+
     guard_script_from_debounce_data(&ns).await?;
 
     let codebase = ns.codebase.as_ref();

backend/windmill-queue/src/jobs.rs

@@ -1107,15 +1107,16 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
     // Resolve the concurrency-limit settings on the pool *before* opening the
     // completion transaction: doing it inside the tx would hold a second
     // simultaneous connection from the small per-worker pool.
-    let has_concurrent_limit = completed_job.concurrent_limit.is_some()
-        || windmill_common::runnable_settings::prefetch_cached_from_handle(
-            completed_job.runnable_settings_handle,
-            db,
-        )
-        .await?
-        .1
-        .concurrent_limit
-        .is_some();
+    let has_concurrent_limit = has_active_concurrency_limit(completed_job.concurrent_limit)
+        || has_active_concurrency_limit(
+            windmill_common::runnable_settings::prefetch_cached_from_handle(
+                completed_job.runnable_settings_handle,
+                db,
+            )
+            .await?
+            .1
+            .concurrent_limit,
+        );
 
     // A genuine NUL (U+0000) in the result serializes to a `\u0000` escape that
     // the jsonb `result` column rejects with 22P05 ("unsupported Unicode escape
@@ -3921,7 +3922,7 @@ pub async fn pull(
                 let pulled_job_result = match job {
                     #[cfg(feature = "private")]
                     Some(job)
-                        if concurrency_settings.concurrent_limit.is_some()
+                        if has_active_concurrency_limit(concurrency_settings.concurrent_limit)
                             // Concurrency limit is available for either enterprise job or dependency job
                             && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DEBOUNCING)) =>
                     {
@@ -3985,7 +3986,8 @@ pub async fn pull(
         .1
         .maybe_fallback(None, job.concurrent_limit, job.concurrency_time_window_s);
 
-        let has_concurent_limit = concurrency_settings.concurrent_limit.is_some();
+        let has_concurent_limit =
+            has_active_concurrency_limit(concurrency_settings.concurrent_limit);
 
         #[cfg(not(feature = "enterprise"))]
         if has_concurent_limit && !job.is_dependency() {
@@ -3994,7 +3996,7 @@ pub async fn pull(
 
         #[cfg(not(feature = "enterprise"))]
         let has_concurent_limit = job.is_dependency()
-            && job.concurrent_limit.is_some()
+            && has_active_concurrency_limit(job.concurrent_limit)
             && cfg!(feature = "private")
             && !*WMDEBUG_NO_DEBOUNCING;
         // if we don't have private flag, we don't have concurrency limit
@@ -4162,6 +4164,13 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>(
     Ok(job_and_suspended)
 }
 
+/// A concurrency limit is only active when it caps at 1+ slots. `Some(0)` (or negative) is
+/// a disabled limit, not a zero-slot one — see [`ConcurrencySettings::normalized`]. The gate
+/// checks must use this instead of `.is_some()` so a legacy stored `0` behaves as disabled.
+pub fn has_active_concurrency_limit(concurrent_limit: Option<i32>) -> bool {
+    concurrent_limit.is_some_and(|n| n > 0)
+}
+
 pub async fn custom_concurrency_key(
     db: &Pool<Postgres>,
     job_id: &Uuid,
@@ -6141,6 +6150,11 @@ async fn push_inner<'c, 'd>(
         },
     };
 
+    // Guard against an already-stored `concurrent_limit <= 0` reaching the queue: it would
+    // register a zero-slot concurrency key and permanently block the job. Coerce it to
+    // disabled before it is persisted onto the job row / concurrency key here.
+    concurrency_settings = concurrency_settings.normalized();
+
     // Enforce concurrency limit on all dependency jobs.
     // TODO: We can ignore this for scripts djobs. The main reason we need all djobs to be sequential is because we have
     // nodes_to_relock and we need all locks whose corresponding steps aren't in nodes_to_relock be already present.
@@ -6328,7 +6342,7 @@ async fn push_inner<'c, 'd>(
         check_workspace_queue_cap(&mut *tx, workspace_id).await?;
     }
 
-    if concurrency_settings.concurrent_limit.is_some() {
+    if has_active_concurrency_limit(concurrency_settings.concurrent_limit) {
         let concurrency_key = resolve_concurrency_key(
             workspace_id,
             &args,
@@ -6732,7 +6746,7 @@ pub async fn insert_concurrency_key_capped<'d, 'c, E: PgExecutor<'c> + Copy>(
         custom_concurrency_key,
     );
     #[cfg(feature = "cloud")]
-    if *CLOUD_HOSTED && concurrent_limit.is_some() {
+    if *CLOUD_HOSTED && has_active_concurrency_limit(concurrent_limit) {
         check_concurrency_key_queue_cap(db, &concurrency_key).await?;
     }
     #[cfg(not(feature = "cloud"))]

backend/windmill-queue/tests/concurrency_limit_zero_test.rs

@@ -0,0 +1,21 @@
+//! Runtime gate for the `Some(0)` concurrency footgun: a stored `concurrent_limit <= 0`
+//! must read as "disabled", never as a zero-slot cap that permanently blocks the job at the
+//! concurrency gate (the re-queue storm the zombie monitor eventually fails as a fake OOM).
+//!
+//! Run with:
+//!   cargo test -p windmill-queue --test concurrency_limit_zero_test
+
+use windmill_queue::jobs::has_active_concurrency_limit;
+
+#[test]
+fn zero_and_negative_are_not_active_limits() {
+    assert!(!has_active_concurrency_limit(None));
+    assert!(!has_active_concurrency_limit(Some(0)));
+    assert!(!has_active_concurrency_limit(Some(-1)));
+}
+
+#[test]
+fn positive_limit_is_active() {
+    assert!(has_active_concurrency_limit(Some(1)));
+    assert!(has_active_concurrency_limit(Some(i32::MAX)));
+}

backend/windmill-types/src/runnable_settings.rs

@@ -93,9 +93,7 @@ pub struct DebouncingSettings {
     pub debounce_args_to_accumulate: Option<Vec<String>>,
 }
 
-#[derive(
-    Debug, Default, Clone, Serialize, Deserialize, Hash, PartialEq, sqlx::FromRow, sqlx::Decode,
-)]
+#[derive(Debug, Default, Clone, Serialize, Hash, PartialEq, sqlx::FromRow, sqlx::Decode)]
 pub struct ConcurrencySettings {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub concurrency_key: Option<String>,
@@ -105,7 +103,65 @@ pub struct ConcurrencySettings {
     pub concurrency_time_window_s: Option<i32>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)]
+/// Shared normalization for the positive-only `Option<i32>` runnable settings
+/// (`concurrent_limit`, `timeout`, ...): a `<= 0` value is never meaningful — zero
+/// concurrent slots permanently blocks a runnable at the concurrency gate (a re-queue
+/// storm the zombie monitor eventually fails with a misleading OOM error), and a
+/// 0-second timeout kills every job on the spot. The frontend already treats `0` as
+/// "disabled", so `<= 0` maps to `None` (unset) everywhere. Idempotent.
+pub fn none_if_non_positive(v: Option<i32>) -> Option<i32> {
+    v.filter(|n| *n > 0)
+}
+
+/// Coerce a `concurrent_limit <= 0` to disabled, dropping the now-meaningless time window
+/// alongside it. Idempotent.
+fn normalize_concurrency(
+    concurrent_limit: &mut Option<i32>,
+    concurrency_time_window_s: &mut Option<i32>,
+) {
+    if none_if_non_positive(*concurrent_limit).is_none() {
+        *concurrent_limit = None;
+        *concurrency_time_window_s = None;
+    }
+}
+
+impl ConcurrencySettings {
+    pub fn normalized(mut self) -> Self {
+        normalize_concurrency(
+            &mut self.concurrent_limit,
+            &mut self.concurrency_time_window_s,
+        );
+        self
+    }
+}
+
+// Manual `Deserialize` so every ingestion path (script/flow create & update, app and
+// http-trigger payloads, and read-back of already-stored settings) normalizes a `<= 0`
+// limit uniformly, without each call site remembering to call `normalized()`.
+impl<'de> Deserialize<'de> for ConcurrencySettings {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Raw {
+            #[serde(default)]
+            concurrency_key: Option<String>,
+            #[serde(default)]
+            concurrent_limit: Option<i32>,
+            #[serde(default)]
+            concurrency_time_window_s: Option<i32>,
+        }
+        let Raw { concurrency_key, concurrent_limit, concurrency_time_window_s } =
+            Raw::deserialize(deserializer)?;
+        Ok(
+            ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s }
+                .normalized(),
+        )
+    }
+}
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow, Default)]
 pub struct ConcurrencySettingsWithCustom {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub custom_concurrency_key: Option<String>,
@@ -115,6 +171,41 @@ pub struct ConcurrencySettingsWithCustom {
     pub concurrency_time_window_s: Option<i32>,
 }
 
+impl ConcurrencySettingsWithCustom {
+    pub fn normalized(mut self) -> Self {
+        normalize_concurrency(
+            &mut self.concurrent_limit,
+            &mut self.concurrency_time_window_s,
+        );
+        self
+    }
+}
+
+impl<'de> Deserialize<'de> for ConcurrencySettingsWithCustom {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Raw {
+            #[serde(default)]
+            custom_concurrency_key: Option<String>,
+            #[serde(default)]
+            concurrent_limit: Option<i32>,
+            #[serde(default)]
+            concurrency_time_window_s: Option<i32>,
+        }
+        let Raw { custom_concurrency_key, concurrent_limit, concurrency_time_window_s } =
+            Raw::deserialize(deserializer)?;
+        Ok(ConcurrencySettingsWithCustom {
+            custom_concurrency_key,
+            concurrent_limit,
+            concurrency_time_window_s,
+        }
+        .normalized())
+    }
+}
+
 impl DebouncingSettings {
     pub fn maybe_fallback(
         self,
@@ -142,11 +233,15 @@ impl ConcurrencySettings {
         concurrent_limit: Option<i32>,
         concurrency_time_window_s: Option<i32>,
     ) -> Self {
+        // Legacy columns can still hold a stored `0` that predates ingestion normalization,
+        // so re-normalize here: this is the single load boundary for every DB-backed read
+        // (script/schedule read, flow value, and the worker pull path).
         Self {
             concurrency_key: self.concurrency_key.or(concurrency_key),
             concurrent_limit: self.concurrent_limit.or(concurrent_limit),
             concurrency_time_window_s: self.concurrency_time_window_s.or(concurrency_time_window_s),
         }
+        .normalized()
     }
 }
 
@@ -229,4 +324,91 @@ mod tests {
         assert_eq!(r, Retry::default());
         assert_eq!(r.exponential.multiplier, 1);
     }
+
+    // The positive-only settings share one rule: `<= 0` means "unset". This is what keeps a
+    // stored `0` from being enforced as a zero-slot cap or a 0-second timeout.
+    #[test]
+    fn none_if_non_positive_coerces_zero_and_negative() {
+        assert_eq!(none_if_non_positive(Some(0)), None);
+        assert_eq!(none_if_non_positive(Some(-3)), None);
+        assert_eq!(none_if_non_positive(Some(1)), Some(1));
+        assert_eq!(none_if_non_positive(Some(i32::MAX)), Some(i32::MAX));
+        assert_eq!(none_if_non_positive(None), None);
+    }
+
+    // Ingestion path (scripts flatten this on `NewScript`, flows on `FlowModule`): a `0`
+    // concurrent_limit deserializes to disabled and drops the now-meaningless time window,
+    // while a real limit and its window survive untouched.
+    #[test]
+    fn concurrency_settings_deserialize_normalizes_non_positive_limit() {
+        let zero: ConcurrencySettings = serde_json::from_value(
+            serde_json::json!({"concurrent_limit": 0, "concurrency_time_window_s": 30}),
+        )
+        .unwrap();
+        assert_eq!(zero.concurrent_limit, None);
+        assert_eq!(zero.concurrency_time_window_s, None);
+
+        let negative: ConcurrencySettings =
+            serde_json::from_value(serde_json::json!({"concurrent_limit": -1})).unwrap();
+        assert_eq!(negative.concurrent_limit, None);
+
+        let real: ConcurrencySettings = serde_json::from_value(
+            serde_json::json!({"concurrent_limit": 2, "concurrency_time_window_s": 30}),
+        )
+        .unwrap();
+        assert_eq!(real.concurrent_limit, Some(2));
+        assert_eq!(real.concurrency_time_window_s, Some(30));
+    }
+
+    // Per-flow-step overrides use the `custom_concurrency_key` variant; same rule.
+    #[test]
+    fn concurrency_settings_with_custom_deserialize_normalizes() {
+        let zero: ConcurrencySettingsWithCustom = serde_json::from_value(
+            serde_json::json!({"concurrent_limit": 0, "concurrency_time_window_s": 5}),
+        )
+        .unwrap();
+        assert_eq!(zero.concurrent_limit, None);
+        assert_eq!(zero.concurrency_time_window_s, None);
+    }
+
+    // A normalized value serializes with the limit omitted (skip_serializing_if), matching the
+    // frontend's "disabled" representation instead of re-emitting a `0`.
+    #[test]
+    fn normalized_disabled_limit_serializes_as_omitted() {
+        let s =
+            ConcurrencySettings { concurrent_limit: Some(0), ..Default::default() }.normalized();
+        let json = serde_json::to_value(&s).unwrap();
+        assert!(json.get("concurrent_limit").is_none());
+    }
+
+    // Runtime load boundary: legacy rows still hold a raw `0` in the fallback columns. The
+    // fallback must not resurrect it as an active limit.
+    #[test]
+    fn maybe_fallback_normalizes_legacy_zero_column() {
+        let merged = ConcurrencySettings::default().maybe_fallback(None, Some(0), Some(30));
+        assert_eq!(merged.concurrent_limit, None);
+        assert_eq!(merged.concurrency_time_window_s, None);
+    }
+
+    // `NewScript`/`FlowModule` embed the settings via `#[serde(flatten)]`, which drives the
+    // manual Deserialize through a content-buffer deserializer rather than a plain map. Guard
+    // that path: normalization must still fire and sibling fields must still parse.
+    #[test]
+    fn flattened_concurrency_normalizes_and_preserves_siblings() {
+        #[derive(Deserialize)]
+        struct Wrapper {
+            name: String,
+            #[serde(flatten)]
+            concurrency: ConcurrencySettings,
+        }
+        let w: Wrapper = serde_json::from_value(serde_json::json!({
+            "name": "s",
+            "concurrent_limit": 0,
+            "concurrency_time_window_s": 42,
+        }))
+        .unwrap();
+        assert_eq!(w.name, "s");
+        assert_eq!(w.concurrency.concurrent_limit, None);
+        assert_eq!(w.concurrency.concurrency_time_window_s, None);
+    }
 }

backend/windmill-worker/src/common.rs

@@ -1081,7 +1081,9 @@ pub async fn resolve_job_timeout(
         *MAX_TIMEOUT_DURATION
     };
 
-    match custom_timeout_secs {
+    // A `custom_timeout_secs <= 0` is not a 0-second limit but "unset": fall through to the
+    // default/global-max timeout instead of killing the job immediately.
+    match windmill_common::runnable_settings::none_if_non_positive(custom_timeout_secs) {
         Some(timeout_secs)
             if Duration::from_secs(timeout_secs as u64) < global_max_timeout_duration =>
         {

backend/windmill-worker/src/worker.rs

@@ -3892,15 +3892,16 @@ pub async fn handle_queued_job(
 
         #[cfg(not(feature = "enterprise"))]
         if let Connection::Sql(db) = conn {
-            if (job.concurrent_limit.is_some()
-                || windmill_common::runnable_settings::prefetch_cached_from_handle(
-                    job.runnable_settings_handle,
-                    db,
-                )
-                .await?
-                .1
-                .concurrent_limit
-                .is_some())
+            if (windmill_queue::jobs::has_active_concurrency_limit(job.concurrent_limit)
+                || windmill_queue::jobs::has_active_concurrency_limit(
+                    windmill_common::runnable_settings::prefetch_cached_from_handle(
+                        job.runnable_settings_handle,
+                        db,
+                    )
+                    .await?
+                    .1
+                    .concurrent_limit,
+                ))
                 && !job.kind.is_dependency()
             {
                 logs.push_str("---\n");

backend/windmill-worker/src/worker_flow.rs

@@ -1525,9 +1525,14 @@ pub async fn update_flow_status_after_job_completion_internal(
             let concurrency_key = tag_and_concurrency_key
                 .as_ref()
                 .and_then(|x| x.concurrency_key.clone());
-            let concurrent_limit = tag_and_concurrency_key
-                .as_ref()
-                .and_then(|x| x.concurrent_limit);
+            // `concurrent_limit` here can come straight from the raw flow JSON (see
+            // get_tag_and_concurrency), bypassing the ConcurrencySettings deserialization guard,
+            // so a stored `0` must still be coerced to disabled before we register a key for it.
+            let concurrent_limit = windmill_common::runnable_settings::none_if_non_positive(
+                tag_and_concurrency_key
+                    .as_ref()
+                    .and_then(|x| x.concurrent_limit),
+            );
             let concurrency_time_window_s = tag_and_concurrency_key
                 .as_ref()
                 .and_then(|x| x.concurrency_time_window_s);
@@ -4389,13 +4394,10 @@ async fn push_next_flow_job(
             )
             .await?;
 
-            if timeout_value < 0 {
-                return Err(Error::ExecutionErr(
-                    "Timeout value cannot be negative".to_string(),
-                ));
-            }
-
-            Some(timeout_value)
+            // A `<= 0` step timeout (including a negative eval) means "no override": fall back
+            // to the referenced runnable's own timeout rather than a 0-second/negative timeout
+            // that would kill the step instantly.
+            effective_flow_step_timeout(Some(timeout_value), payload_tag.timeout)
         } else {
             payload_tag.timeout
         };
@@ -6048,6 +6050,18 @@ async fn flow_to_payload(
     })
 }
 
+/// Effective timeout for a flow step given the module's (already-evaluated) timeout override and
+/// the timeout inherited from the referenced runnable. A `<= 0` override — or none — means "no
+/// override": fall back to the inherited value (which is itself `None` when unset, i.e. the
+/// instance default). A positive override wins. This keeps a step `timeout: 0` equivalent to an
+/// omitted one rather than a 0-second, instant-kill timeout.
+pub(crate) fn effective_flow_step_timeout(
+    module_override: Option<i32>,
+    inherited: Option<i32>,
+) -> Option<i32> {
+    windmill_common::runnable_settings::none_if_non_positive(module_override).or(inherited)
+}
+
 pub async fn script_to_payload(
     script_hash: Option<windmill_common::scripts::ScriptHash>,
     script_path: String,
@@ -6137,11 +6151,13 @@ pub async fn script_to_payload(
         module.delete_after_use.unwrap_or(false) || delete_after_use.unwrap_or(false);
     let final_delete_after_secs = module.delete_after_secs.or(delete_after_secs);
 
-    let flow_step_timeout = if module.timeout.is_some() {
-        None
-    } else {
-        script_timeout
-    };
+    // Always carry the referenced script's own timeout as the inherited fallback. The module's
+    // timeout override (if any) is selected at the push site, where a `<= 0` override is treated
+    // as "no override" and falls back to this value — so `timeout: 0` on a step means "use the
+    // script's timeout", not a 0-second (immediate-kill) timeout. Normalize the inherited value
+    // too, so a legacy `0` script timeout resolves to the default rather than a zero-second kill.
+    let flow_step_timeout =
+        windmill_common::runnable_settings::none_if_non_positive(script_timeout);
     Ok(JobPayloadWithTag {
         payload,
         tag,
@@ -6266,9 +6282,25 @@ pub async fn get_previous_job_result(
 
 #[cfg(test)]
 mod tests {
-    use super::extract_chat_message_from_flow_result;
+    use super::{effective_flow_step_timeout, extract_chat_message_from_flow_result};
     use serde_json::{json, value::to_raw_value};
 
+    // A `<= 0` step timeout override must behave as "no override" and inherit the referenced
+    // script's timeout, not collapse to a 0-second (instant-kill) timeout. A positive override
+    // still wins. Guards the flow-step timeout footgun.
+    #[test]
+    fn flow_step_timeout_zero_or_negative_inherits_script_timeout() {
+        // zero / negative override -> inherited script timeout
+        assert_eq!(effective_flow_step_timeout(Some(0), Some(300)), Some(300));
+        assert_eq!(effective_flow_step_timeout(Some(-5), Some(300)), Some(300));
+        // no inherited timeout either -> None (falls through to the instance default)
+        assert_eq!(effective_flow_step_timeout(Some(0), None), None);
+        // positive override wins over the inherited value
+        assert_eq!(effective_flow_step_timeout(Some(120), Some(300)), Some(120));
+        // no override -> inherited
+        assert_eq!(effective_flow_step_timeout(None, Some(300)), Some(300));
+    }
+
     #[test]
     fn pretty_prints_full_result_when_no_override_is_present() {
         let value = json!({

cli/src/commands/script/script.ts

@@ -100,6 +100,30 @@ export function isRawAppBackendPath(filePath: string): boolean {
   return isRawAppBackendPathInternal(filePath);
 }
 
+/**
+ * The positive-only runnable settings (concurrent_limit, timeout, ...) treat any `<= 0`
+ * value as "unset": the backend coerces it to null (a 0-slot concurrency limit bricks the
+ * runnable, a 0s timeout kills every run). Coerce to undefined so it is serialized as
+ * omitted, never as 0, and redeploys don't churn against the backend-normalized value.
+ */
+export function nonePositiveInt(
+  v: number | undefined | null
+): number | undefined {
+  return v != null && v > 0 ? v : undefined;
+}
+
+/**
+ * Normalize a concurrent_limit + its time window together: when the limit is disabled
+ * (<= 0) the window is dropped too. Returns [concurrent_limit, concurrency_time_window_s].
+ */
+export function normalizeConcurrency(
+  concurrentLimit: number | undefined | null,
+  concurrencyTimeWindowS?: number | undefined | null
+): [number | undefined, number | undefined] {
+  const limit = nonePositiveInt(concurrentLimit);
+  return limit === undefined ? [undefined, undefined] : [limit, concurrencyTimeWindowS ?? undefined];
+}
+
 /**
  * Checks if a path is inside a normal app folder (inline script).
  * Matches patterns like: .../myApp.app/... or .../myApp__app/...
@@ -469,6 +493,15 @@ export async function handleFile(
     const moduleFolderPath = scriptBasePath + getModuleFolderSuffix();
     const modules = await readModulesFromDisk(moduleFolderPath, opts?.defaultTs, moduleEntryPoint);
 
+    // A concurrent_limit of <= 0 means "concurrency disabled", not "zero slots" (which
+    // would brick the runnable at the queue's concurrency gate). Emit it as omitted rather
+    // than 0 so a redeploy never re-persists a zero-slot limit, and drop the now-meaningless
+    // time window alongside it. Mirrors the backend's ConcurrencySettings::normalized.
+    const [normConcurrentLimit, normConcurrencyTimeWindowS] = normalizeConcurrency(
+      typed?.concurrent_limit,
+      typed?.concurrency_time_window_s
+    );
+
     const requestBodyCommon: NewScript = {
       content,
       description: typed?.description ?? "",
@@ -482,8 +515,8 @@ export async function handleFile(
       ws_error_handler_muted: typed?.ws_error_handler_muted,
       dedicated_worker: typed?.dedicated_worker,
       cache_ttl: typed?.cache_ttl,
-      concurrency_time_window_s: typed?.concurrency_time_window_s,
-      concurrent_limit: typed?.concurrent_limit,
+      concurrency_time_window_s: normConcurrencyTimeWindowS,
+      concurrent_limit: normConcurrentLimit,
       deployment_message: message,
       restart_unless_cancelled: typed?.restart_unless_cancelled,
       visible_to_runner_only: typed?.visible_to_runner_only,
@@ -493,7 +526,7 @@ export async function handleFile(
       debounce_key: typed?.debounce_key,
       debounce_delay_s: typed?.debounce_delay_s,
       codebase: await codebase?.getDigest(forceTar),
-      timeout: typed?.timeout,
+      timeout: nonePositiveInt(typed?.timeout),
       on_behalf_of_email: typed?.on_behalf_of_email,
       envs: typed?.envs,
       modules: modules,
@@ -530,9 +563,13 @@ export async function handleFile(
               remote.ws_error_handler_muted &&
             typed.dedicated_worker == remote.dedicated_worker &&
             typed.cache_ttl == remote.cache_ttl &&
-            typed.concurrency_time_window_s ==
-              remote.concurrency_time_window_s &&
-            typed.concurrent_limit == remote.concurrent_limit &&
+            normConcurrencyTimeWindowS ==
+              normalizeConcurrency(
+                remote.concurrent_limit,
+                remote.concurrency_time_window_s
+              )[1] &&
+            normConcurrentLimit ==
+              normalizeConcurrency(remote.concurrent_limit)[0] &&
             Boolean(typed.restart_unless_cancelled) ==
               Boolean(remote.restart_unless_cancelled) &&
             Boolean(typed.visible_to_runner_only) ==
@@ -540,7 +577,7 @@ export async function handleFile(
             Boolean(typed.has_preprocessor) ==
               Boolean(remote.has_preprocessor) &&
             typed.priority == Boolean(remote.priority) &&
-            typed.timeout == remote.timeout &&
+            nonePositiveInt(typed.timeout) == nonePositiveInt(remote.timeout) &&
             //@ts-ignore
             typed.concurrency_key == remote["concurrency_key"] &&
             typed.debounce_key == remote["debounce_key"] &&