diff --git a/backend/.sqlx/query-903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625.json b/backend/.sqlx/query-456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47.json
similarity index 56%
rename from backend/.sqlx/query-903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625.json
rename to backend/.sqlx/query-456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47.json
index abef34ebc0..c3dd9208ba 100644
--- a/backend/.sqlx/query-903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625.json
+++ b/backend/.sqlx/query-456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n                         FROM v2_job WHERE id = $1",
+  "query": "\n            SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n            FROM v2_job\n            WHERE id = $1\n            ",
   "describe": {
     "columns": [
       {
@@ -18,5 +18,5 @@
       true
     ]
   },
-  "hash": "903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625"
+  "hash": "456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47"
 }
diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
index 713ccb9dd3..36ddb8ab9f 100644
--- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
+++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
@@ -15,7 +15,7 @@
       ]
     },
     "nullable": [
-      null
+      true
     ]
   },
   "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
diff --git a/backend/.sqlx/query-b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149.json b/backend/.sqlx/query-b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149.json
deleted file mode 100644
index f37bea0531..0000000000
--- a/backend/.sqlx/query-b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149.json
+++ /dev/null
@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n                                 args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n                             FROM v2_job\n                             WHERE id = $1",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "args: Json<HashMap<String, Box<RawValue>>>",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": [
-      true
-    ]
-  },
-  "hash": "b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149"
-}
diff --git a/backend/Cargo.lock b/backend/Cargo.lock
index 959cd8bf84..c5b0e48db7 100644
--- a/backend/Cargo.lock
+++ b/backend/Cargo.lock
@@ -15753,6 +15753,7 @@ name = "windmill-worker"
 version = "1.538.0"
 dependencies = [
  "anyhow",
+ "async-once-cell",
  "async-recursion",
  "async-stream",
  "backon",
diff --git a/backend/Cargo.toml b/backend/Cargo.toml
index 661e932a96..bbb04451e6 100644
--- a/backend/Cargo.toml
+++ b/backend/Cargo.toml
@@ -367,7 +367,7 @@ aws-sdk-sts = "=1.79.0"
 aws-sdk-sso = "=1.77.0"
 aws-sdk-ssooidc = "=1.78.0"
 rustls = "=0.23.29"
-
+async-once-cell = "0.5.4"
 
 
 aws-smithy-types-convert = { version = "^0", features = ["convert-chrono"] }
diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs
index 69bb707ad7..c8f861b561 100644
--- a/backend/windmill-api/src/flows.rs
+++ b/backend/windmill-api/src/flows.rs
@@ -1653,7 +1653,8 @@ mod tests {
                   "exponential": {
                     "multiplier": 1,
                     "seconds": 0
-                  }
+                  },
+                  "retry_if": null
                 }
                 "#
             )
@@ -1668,13 +1669,15 @@ mod tests {
                     multiplier: 1,
                     seconds: 123,
                     random_factor: None
-                }
+                },
+                retry_if: None
             },
             serde_json::from_str(
                 r#"
                 {
                   "constant": {},
-                  "exponential": { "seconds": 123 }
+                  "exponential": { "seconds": 123 },
+                  "retry_if" : null
                 }
                 "#
             )
@@ -1692,6 +1695,7 @@ mod tests {
                 seconds: 3,
                 random_factor: None,
             },
+            retry_if: None,
         };
         assert_eq!(
             vec![
@@ -1718,6 +1722,7 @@ mod tests {
                 seconds: 3,
                 random_factor: None,
             },
+            retry_if: None,
         };
         assert_eq!(
             vec![
diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs
index 5a905c65ca..1be402612b 100644
--- a/backend/windmill-common/src/flows.rs
+++ b/backend/windmill-common/src/flows.rs
@@ -12,6 +12,7 @@ use std::{
     u8,
 };
 
+use anyhow::Context;
 use rand::Rng;
 use serde::{Deserialize, Serialize, Serializer};
 use sqlx::types::Json;
@@ -137,6 +138,65 @@ pub struct FlowValue {
     pub concurrency_key: Option<String>,
 }
 
+impl FlowValue {
+    pub fn get_flow_module_at_step(&self, step: Step) -> anyhow::Result<&FlowModule> {
+        let flow_module = match step {
+            Step::PreprocessorStep => self
+                .preprocessor_module
+                .as_deref()
+                .with_context(|| format!("no preprocessor module")),
+            Step::Step(i) => self
+                .modules
+                .get(i)
+                .with_context(|| format!("no module found at index: {i}")),
+            Step::FailureStep => self
+                .failure_module
+                .as_deref()
+                .with_context(|| format!("no failure module")),
+        };
+
+        flow_module
+    }
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum Step {
+    Step(usize),
+    PreprocessorStep,
+    FailureStep,
+}
+
+impl Step {
+    pub fn from_i32_and_len(step: i32, len: usize) -> Self {
+        if step < 0 {
+            Step::PreprocessorStep
+        } else if (step as usize) < len {
+            Step::Step(step as usize)
+        } else {
+            Step::FailureStep
+        }
+    }
+
+    pub fn get_step_index(&self) -> Option<usize> {
+        match self {
+            Step::Step(index) => Some(*index),
+            _ => None,
+        }
+    }
+
+    pub fn is_index_step(&self) -> bool {
+        matches!(self, Step::Step(_))
+    }
+
+    pub fn is_preprocessor_step(&self) -> bool {
+        matches!(self, Step::PreprocessorStep)
+    }
+
+    pub fn is_failure_step(&self) -> bool {
+        matches!(self, Step::FailureStep)
+    }
+}
+
 #[derive(Default, Deserialize, Serialize, Debug, Clone)]
 pub struct StopAfterIf {
     pub expr: String,
@@ -144,11 +204,18 @@ pub struct StopAfterIf {
     pub error_message: Option<String>,
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
+pub struct RetryIf {
+    pub expr: String,
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
 #[serde(default)]
 pub struct Retry {
     pub constant: ConstantDelay,
     pub exponential: ExponentialDelay,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub retry_if: Option<RetryIf>,
 }
 
 impl Retry {
@@ -156,7 +223,7 @@ impl Retry {
     ///
     /// May return [`Duration::ZERO`] to retry immediately.
     pub fn interval(&self, previous_attempts: u32, silent: bool) -> Option<Duration> {
-        let Self { constant, exponential } = self;
+        let Self { constant, exponential, .. } = self;
 
         if previous_attempts < constant.attempts {
             Some(Duration::from_secs(constant.seconds as u64))
diff --git a/backend/windmill-queue/src/flow_status.rs b/backend/windmill-queue/src/flow_status.rs
index 165a161e65..50af234149 100644
--- a/backend/windmill-queue/src/flow_status.rs
+++ b/backend/windmill-queue/src/flow_status.rs
@@ -1,29 +1,8 @@
 use uuid::Uuid;
 use windmill_common::{
-    error::{self, Error},
-    utils::WarnAfterExt,
-    DB,
+    error::{self, Error}, flows::Step, utils::WarnAfterExt, DB
 };
 
-#[derive(Debug, Copy, Clone)]
-pub enum Step {
-    Step(usize),
-    PreprocessorStep,
-    FailureStep,
-}
-
-impl Step {
-    pub fn from_i32_and_len(step: i32, len: usize) -> Self {
-        if step < 0 {
-            Step::PreprocessorStep
-        } else if (step as usize) < len {
-            Step::Step(step as usize)
-        } else {
-            Step::FailureStep
-        }
-    }
-}
-
 pub async fn update_flow_status_in_progress(
     db: &DB,
     _w_id: &str,
diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml
index bbb1b414ed..b285224ed4 100644
--- a/backend/windmill-worker/Cargo.toml
+++ b/backend/windmill-worker/Cargo.toml
@@ -124,6 +124,7 @@ backon.workspace = true
 winapi = { workspace = true, optional = true }
 pep440_rs.workspace = true
 process-wrap.workspace = true
+async-once-cell.workspace = true
 
 opentelemetry = { workspace = true, optional = true }
 bollard = { workspace = true, optional = true }
diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs
index 11ec4d7069..d196b5f3cd 100644
--- a/backend/windmill-worker/src/ai_executor.rs
+++ b/backend/windmill-worker/src/ai_executor.rs
@@ -12,7 +12,7 @@ use windmill_common::{
     db::DB,
     error::{self, to_anyhow, Error},
     flow_status::AgentAction,
-    flows::{FlowModule, FlowModuleValue},
+    flows::{FlowModule, FlowModuleValue, Step},
     get_latest_hash_for_path,
     jobs::JobKind,
     scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
@@ -21,9 +21,8 @@ use windmill_common::{
 };
 use windmill_parser::Typ;
 use windmill_queue::{
-    flow_status::{get_step_of_flow_status, Step},
-    get_mini_pulled_job, push, CanceledBy, JobCompleted, MiniPulledJob, PushArgs,
-    PushIsolationLevel,
+    flow_status::get_step_of_flow_status, get_mini_pulled_job, push, CanceledBy, JobCompleted,
+    MiniPulledJob, PushArgs, PushIsolationLevel,
 };
 
 use crate::{
diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 405ed49307..92c55f103c 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -20,6 +20,7 @@ use crate::{
 };
 
 use anyhow::Context;
+use async_once_cell::Lazy;
 use futures::TryFutureExt;
 use mappable_rc::Marc;
 use serde::{Deserialize, Serialize};
@@ -56,9 +57,8 @@ use windmill_common::{
         Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
         MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
-    flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
+    flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Step, Suspend},
 };
-use windmill_queue::flow_status::Step;
 use windmill_queue::schedule::get_schedule_opt;
 use windmill_queue::{
     add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
@@ -222,6 +222,7 @@ fn result_has_recover_true(nresult: Arc<Box<RawValue>>) -> bool {
     let recover = serde_json::from_str::<RecoveryObject>(nresult.get());
     return recover.map(|r| r.recover.unwrap_or(false)).unwrap_or(false);
 }
+
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion_internal(
     db: &DB,
@@ -344,14 +345,33 @@ pub async fn update_flow_status_after_job_completion_internal(
         let is_failure_step =
             old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;
 
+        let args = Arc::pin(Lazy::new(async move {
+            let args = sqlx::query_scalar!(
+                r#"
+            SELECT args AS "args: Json<HashMap<String, Box<RawValue>>>"
+            FROM v2_job
+            WHERE id = $1
+            "#,
+                flow
+            )
+            .fetch_one(db)
+            .await;
+            args
+        }));
+
+        let from_result_to_args =
+            |args: &Result<Option<Json<HashMap<String, Box<RawValue>>>>, sqlx::Error>| {
+                let args = args.as_ref().map_err(|e| {
+                    Error::internal_err(format!("retrieval of args from state: {e:#}"))
+                })?;
+
+                Ok::<_, Error>(args.clone())
+            };
+
         let (mut stop_early, mut stop_early_err_msg, mut skip_if_stop_early, continue_on_error) =
             if let Some(se) = stop_early_override {
                 //do not stop early if module is a flow step
-                let step = match module_step {
-                    Step::PreprocessorStep => None,
-                    Step::FailureStep => None,
-                    Step::Step(i) => Some(i),
-                };
+                let step = module_step.get_step_index();
 
                 let is_flow = if let Some(_) = step {
                     #[derive(Deserialize)]
@@ -375,7 +395,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                 } else {
                     (true, None, se, false)
                 }
-            } else if is_failure_step || matches!(module_step, Step::PreprocessorStep) {
+            } else if is_failure_step || module_step.is_preprocessor_step() {
                 (false, None, false, false)
             } else if let Some(current_module) = current_module {
                 let stop_early = success
@@ -394,18 +414,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                                 )),
                                 _ => None,
                             };
-                        let args = sqlx::query_scalar!(
-                            "SELECT
-                                 args AS \"args: Json<HashMap<String, Box<RawValue>>>\"
-                             FROM v2_job
-                             WHERE id = $1",
-                            flow
-                        )
-                        .fetch_one(db)
-                        .await
-                        .map_err(|e| {
-                            Error::internal_err(format!("retrieval of args from state: {e:#}"))
-                        })?;
+                        let args = from_result_to_args(args.as_ref().await.get_ref())?;
                         compute_bool_from_expr(
                             &expr,
                             Marc::new(args.unwrap_or_default().0),
@@ -820,7 +829,17 @@ pub async fn update_flow_status_after_job_completion_internal(
                             .unwrap_or_default();
 
                         tracing::info!("update flow status on retry: {retry:#?} ");
-                        next_retry(&retry, &old_status.retry).is_none()
+                        let args = from_result_to_args(args.as_ref().await.get_ref())?;
+
+                        evaluate_retry(
+                            &retry,
+                            &old_status.retry,
+                            result.clone(),
+                            Marc::new(args.unwrap_or_default().0),
+                            Some(client),
+                        )
+                        .await?
+                        .is_none()
                     } else {
                         false
                     };
@@ -901,7 +920,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                         "error while setting flow status in failure step: {e:#}"
                     ))
                 })?;
-            } else if matches!(module_step, Step::PreprocessorStep) {
+            } else if module_step.is_preprocessor_step() {
                 sqlx::query!(
                     "UPDATE v2_job_status
                      SET flow_status = JSONB_SET(flow_status, ARRAY['preprocessor_module'], $1)
@@ -968,16 +987,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                     .as_ref()
                     .and_then(|m| m.stop_after_all_iters_if.as_ref())
                 {
-                    let args = sqlx::query_scalar!(
-                        "SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"
-                         FROM v2_job WHERE id = $1",
-                        flow
-                    )
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::internal_err(format!("retrieval of args from state: {e:#}"))
-                    })?;
+                    let args = from_result_to_args(args.as_ref().await.get_ref())?;
 
                     let should_stop = compute_bool_from_expr(
                         &stop_after_all_iters_if.expr,
@@ -1025,7 +1035,7 @@ pub async fn update_flow_status_after_job_completion_internal(
             .ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?;
         tx.commit().await?;
 
-        if matches!(module_step, Step::PreprocessorStep) && success {
+        if module_step.is_preprocessor_step() && success {
             let tag_and_concurrency_key = get_tag_and_concurrency(&flow, db).await;
             let require_args = tag_and_concurrency_key.as_ref().is_some_and(|x| {
                 x.tag.as_ref().is_some_and(|t| t.contains("$args"))
@@ -1150,6 +1160,28 @@ pub async fn update_flow_status_after_job_completion_internal(
             .map(|x| x.to_string())
             .unwrap_or_else(|| "none".to_string());
 
+        let should_retry = async move || -> error::Result<bool> {
+            let default_retry = Retry::default();
+            let retry_config = flow_value
+                .get_flow_module_at_step(module_step)
+                .ok()
+                .and_then(|flow_module| flow_module.retry.as_ref())
+                .unwrap_or(&default_retry);
+            let args = from_result_to_args(args.as_ref().await.get_ref())?;
+
+            let should_retry = evaluate_retry(
+                retry_config,
+                &old_status.retry,
+                result.clone(),
+                Marc::new(args.unwrap_or_default().0),
+                Some(client),
+            )
+            .await?
+            .is_some();
+
+            Ok(should_retry)
+        };
+
         let should_continue_flow = match success {
             _ if stop_early => false,
             _ if flow_job.is_canceled() => false,
@@ -1158,30 +1190,7 @@ pub async fn update_flow_status_after_job_completion_internal(
             false if skip_seq_branch_failure || skip_loop_failures || continue_on_error => {
                 !is_last_step
             }
-            false
-                if next_retry(
-                    match module_step {
-                        Step::PreprocessorStep => flow_value
-                            .preprocessor_module
-                            .as_ref()
-                            .and_then(|m| m.retry.as_ref()),
-                        Step::Step(i) => flow_value
-                            .modules
-                            .get(i)
-                            .as_ref()
-                            .and_then(|m| m.retry.as_ref()),
-                        Step::FailureStep => flow_value
-                            .failure_module
-                            .as_ref()
-                            .and_then(|m| m.retry.as_ref()),
-                    }
-                    .unwrap_or(&Retry::default()),
-                    &old_status.retry,
-                )
-                .is_some() =>
-            {
-                true
-            }
+            false if should_retry().await? => true,
             false
                 if !is_failure_step
                     && !has_triggered_error_handler
@@ -1532,11 +1541,39 @@ async fn compute_skip_branchall_failure<'c>(
 //         )))
 // }
 
-fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u32, Duration)> {
-    (status.fail_count <= MAX_RETRY_ATTEMPTS)
-        .then(|| &retry)
-        .and_then(|retry| retry.interval(status.fail_count, false))
-        .map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL)))
+async fn evaluate_retry(
+    retry: &Retry,
+    status: &RetryStatus,
+    result: Arc<Box<RawValue>>,
+    flow_args: Marc<HashMap<String, Box<RawValue>>>,
+    client: Option<&AuthedClient>,
+) -> anyhow::Result<Option<(u32, Duration)>> {
+    if status.fail_count > MAX_RETRY_ATTEMPTS {
+        return Ok(None);
+    }
+
+    if let Some(retry_if) = &retry.retry_if {
+        let should_retry = compute_bool_from_expr(
+            &retry_if.expr,
+            flow_args,
+            result,
+            None,
+            None,
+            client,
+            None,
+            None,
+        )
+        .await?;
+
+        if !should_retry {
+            tracing::debug!("Retry condition evaluated to false, not retrying");
+            return Ok(None);
+        }
+    }
+
+    Ok(retry
+        .interval(status.fail_count, false)
+        .map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL))))
 }
 
 async fn compute_bool_from_expr(
@@ -1959,7 +1996,7 @@ async fn push_next_flow_job(
     let arc_last_job_result = if status_module.is_failure() {
         // if job is being retried, pass the result of its previous failure
         last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}"))))
-    } else if matches!(step, Step::Step(0)) || matches!(step, Step::PreprocessorStep) {
+    } else if matches!(step, Step::Step(0)) || step.is_preprocessor_step() {
         // if it's the first job executed in the flow, pass the flow args
         Arc::new(to_raw_value(&flow_job.args))
     } else {
@@ -2270,20 +2307,7 @@ async fn push_next_flow_job(
         }
     }
 
-    let mut module = match step {
-        Step::Step(i) => flow
-            .modules
-            .get(i)
-            .with_context(|| format!("no module at index {}", i))?,
-        Step::PreprocessorStep => flow
-            .preprocessor_module
-            .as_ref()
-            .with_context(|| format!("no preprocessor module"))?,
-        Step::FailureStep => flow
-            .failure_module
-            .as_deref()
-            .with_context(|| format!("no failure module"))?,
-    };
+    let mut module = flow.get_flow_module_at_step(step)?;
 
     let current_id = &module.id;
     let mut previous_id = match step {
@@ -2365,7 +2389,14 @@ async fn push_next_flow_job(
 
     let retry = if matches!(&status_module, FlowStatusModule::Failure { .. },) {
         let retry = &module.retry.clone().unwrap_or_default();
-        next_retry(retry, &status.retry)
+        evaluate_retry(
+            retry,
+            &status.retry,
+            arc_last_job_result.clone(),
+            arc_flow_job_args.clone(),
+            Some(client),
+        )
+        .await?
     } else {
         None
     };
@@ -2497,7 +2528,7 @@ async fn push_next_flow_job(
         } else {
             Ok(Marc::new(HashMap::new()))
         }
-    } else if matches!(step, Step::PreprocessorStep) {
+    } else if step.is_preprocessor_step() {
         let mut hm = (*arc_flow_job_args).clone();
         hm.insert(
             ENTRYPOINT_OVERRIDE.to_string(),
@@ -2835,7 +2866,7 @@ async fn push_next_flow_job(
                 .map(|x| x.into());
 
         tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
-        let tag = if !matches!(step, Step::PreprocessorStep)
+        let tag = if !step.is_preprocessor_step()
             && (flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id))
         {
             payload_tag.tag.clone()
diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
index 0e2c4d4a4d..2d180631fe 100644
--- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
@@ -648,7 +648,7 @@
 															</Tooltip>
 														</div>
 														<div class="my-8"></div>
-														<FlowRetries bind:flowModuleRetry={flowModule.retry} />
+														<FlowRetries bind:flowModuleRetry={flowModule.retry} bind:flowModule />
 													</Section>
 												{:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'concurrency'}
 													<Section label="Concurrency limits" class="flex flex-col gap-4" eeOnly>
diff --git a/frontend/src/lib/components/flows/content/FlowRetries.svelte b/frontend/src/lib/components/flows/content/FlowRetries.svelte
index a67ffc2da7..8d14890961 100644
--- a/frontend/src/lib/components/flows/content/FlowRetries.svelte
+++ b/frontend/src/lib/components/flows/content/FlowRetries.svelte
@@ -1,22 +1,64 @@
 <script lang="ts">
-	import type { Retry } from '$lib/gen'
+	import type { Retry, FlowModule } from '$lib/gen'
 	import { SecondsInput } from '$lib/components/common'
 	import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
 	import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
 	import { enterpriseLicense } from '$lib/stores'
 	import { AlertTriangle } from 'lucide-svelte'
-	import { untrack } from 'svelte'
+	import { untrack, getContext } from 'svelte'
+	import Toggle from '$lib/components/Toggle.svelte'
+	import SimpleEditor from '$lib/components/SimpleEditor.svelte'
+	import PropPickerWrapper from '$lib/components/flows/propPicker/PropPickerWrapper.svelte'
+	import Tooltip from '$lib/components/Tooltip.svelte'
+	import Section from '$lib/components/Section.svelte'
+	import type { FlowEditorContext } from '../types'
+	import { getStepPropPicker } from '../previousResults'
+	import { NEVER_TESTED_THIS_FAR } from '../models'
 
 	interface Props {
 		flowModuleRetry: Retry | undefined
 		disabled?: boolean
+		flowModule?: FlowModule
 	}
 
-	let { flowModuleRetry = $bindable(), disabled = false }: Props = $props()
+	let {
+		flowModule = $bindable(),
+		flowModuleRetry = $bindable(),
+		disabled = false
+	}: Props = $props()
+
+	const flowEditorContext = getContext<FlowEditorContext>('FlowEditorContext')
+	const { flowStateStore, flowStore, previewArgs } = flowEditorContext || {
+		flowStateStore: null,
+		flowStore: null,
+		previewArgs: null
+	}
 
 	let delayType = $state() as 'disabled' | 'constant' | 'exponential' | undefined
 	let loaded = $state(false)
 
+	let editor: SimpleEditor | undefined = $state(undefined)
+	let stepPropPicker = $derived(
+		flowModule && flowStateStore?.val && flowStore?.val && previewArgs?.val
+			? getStepPropPicker(
+					flowStateStore.val,
+					undefined,
+					undefined,
+					flowModule.id,
+					flowStore.val,
+					previewArgs.val,
+					false
+				)
+			: null
+	)
+
+	let isRetryConditionEnabled = $derived(Boolean(flowModuleRetry?.retry_if))
+	let result = $derived(
+		flowModule && flowStateStore?.val
+			? (flowStateStore.val[flowModule.id]?.previewResult ?? NEVER_TESTED_THIS_FAR)
+			: NEVER_TESTED_THIS_FAR
+	)
+
 	function setConstantRetries() {
 		flowModuleRetry = {
 			...flowModuleRetry,
@@ -63,7 +105,7 @@
 	const u32Max = 4294967295
 </script>
 
-<div class="h-full flex flex-col">
+<div class="h-full flex flex-col gap-4">
 	<ToggleButtonGroup
 		bind:selected={delayType}
 		class={`h-10 ${disabled ? 'disabled' : ''}`}
@@ -84,6 +126,90 @@
 			<ToggleButton light value="exponential" label="Exponential" {item} />
 		{/snippet}
 	</ToggleButtonGroup>
+
+	{#if delayType === 'constant' || delayType === 'exponential'}
+		<Section label="Retry Condition" class="w-full">
+			{#snippet header()}
+				<Tooltip>
+					Optional condition to determine when to retry. If not specified, will retry on any failure
+					within the configured attempt limits.
+				</Tooltip>
+			{/snippet}
+
+			<Toggle
+				checked={isRetryConditionEnabled}
+				on:change={() => {
+					if (!flowModuleRetry) {
+						return
+					}
+					if (isRetryConditionEnabled && flowModuleRetry.retry_if) {
+						const { retry_if, ...rest } = flowModuleRetry
+						flowModuleRetry = rest
+					} else {
+						flowModuleRetry = {
+							...flowModuleRetry,
+							retry_if: {
+								expr: 'error && error.name !== "PERMANENT_FAILURE"'
+							}
+						}
+					}
+				}}
+				options={{
+					right: 'Only retry if condition is met'
+				}}
+			/>
+
+			<div
+				class="w-full border p-2 mt-2 flex flex-col {flowModuleRetry?.retry_if
+					? ''
+					: 'bg-surface-secondary'}"
+			>
+				{#if flowModuleRetry?.retry_if}
+					<span class="mt-2 text-xs font-bold">Retry condition expression</span>
+					<span class="text-xs text-tertiary mb-2"
+						>Expression should return true to retry, false to skip retry</span
+					>
+					<div class="border w-full">
+						{#if stepPropPicker}
+							<PropPickerWrapper
+								notSelectable
+								pickableProperties={stepPropPicker.pickableProperties}
+								{result}
+								on:select={({ detail }) => {
+									editor?.insertAtCursor(detail)
+									editor?.focus()
+								}}
+							>
+								<SimpleEditor
+									bind:this={editor}
+									lang="javascript"
+									bind:code={flowModuleRetry.retry_if.expr}
+									class="few-lines-editor"
+									extraLib={`declare const result = ${JSON.stringify(result)};` +
+										`\ndeclare const flow_input = ${JSON.stringify(stepPropPicker.pickableProperties.flow_input || {})};`}
+								/>
+							</PropPickerWrapper>
+						{:else}
+							<SimpleEditor
+								bind:this={editor}
+								lang="javascript"
+								bind:code={flowModuleRetry.retry_if.expr}
+								class="few-lines-editor"
+								extraLib={`declare const result = ${JSON.stringify(result)};`}
+							/>
+						{/if}
+					</div>
+				{:else}
+					<span class="mt-2 text-xs font-bold">Retry condition expression</span>
+					<span class="text-xs text-tertiary mb-2"
+						>Expression should return true to retry, false to skip retry</span
+					>
+					<textarea disabled rows="3" class="min-h-[80px]"></textarea>
+				{/if}
+			</div>
+		</Section>
+	{/if}
+
 	<div class="flex h-[calc(100%-22px)]">
 		<div class="w-1/2 h-full overflow-auto pr-2">
 			{#if delayType === 'constant'}
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml
index 5dda8550cc..b0df963c43 100644
--- a/openflow.openapi.yaml
+++ b/openflow.openapi.yaml
@@ -88,6 +88,16 @@ components:
               type: integer
               minimum: 0
               maximum: 100
+        retry_if:
+          $ref: '#/components/schemas/RetryIf'
+
+    RetryIf:
+      type: object
+      properties:
+        expr:
+          type: string
+      required:
+        - expr
 
     StopAfterIf:
       type: object