/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2022
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use crate::common::{cached_result_path, get_root_job_id, save_in_cache, transform_json};
use crate::js_eval::{eval_timeout, IdContext};
use crate::worker_utils::get_tag_and_concurrency;
use crate::{
    JobCompletedSender, PreviousResult, SameWorkerSender, SendResultPayload, UpdateFlow,
    KEEP_JOB_DIR,
};

use anyhow::Context;
use async_once_cell::Lazy;
use backon::{BackoffBuilder, ConstantBuilder, Retryable};
use futures::TryFutureExt;
use mappable_rc::Marc;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use serde_json::{json, Value};
use sqlx::types::Json;
use sqlx::{FromRow, Postgres, Transaction};
use tracing::instrument;
use uuid::Uuid;
use windmill_common::auth::get_job_perms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
use windmill_common::cache::{self, RawData};
use windmill_common::client::AuthedClient;
use windmill_common::db::Authed;
use windmill_common::flow_conversations::{add_message_to_conversation_tx, MessageType};
use windmill_common::flow_status::{
    ApprovalConditions, FlowJobDuration, FlowJobsDuration, FlowStatusModuleWParent,
    Iterator as FlowIterator, JobResult,
};
use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf};
use windmill_common::jobs::{
    check_tag_available_for_workspace_internal, script_path_to_payload, JobKind, JobPayload,
    OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
};
use windmill_common::runnable_settings::{
    ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettingsTrait,
};
use windmill_common::scripts::{ScriptHash, ScriptRunnableSettingsInline};
use windmill_common::users::username_to_permissioned_as;
use windmill_common::utils::WarnAfterExt;
use windmill_common::worker::{error_to_value, to_raw_value, Connection};
use windmill_common::{
    add_time, get_latest_flow_version_info_for_path, get_script_info_for_hash, FlowVersionInfo,
    ScriptHashInfo, DB,
};
use windmill_common::{
    error::{self, to_anyhow, Error},
    flow_status::{
        Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
        MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
    },
    flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Step, Suspend},
    min_version::MIN_VERSION_IS_AT_LEAST_1_595,
};
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
    add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
    insert_concurrency_key, interpolate_args,
    report_error_to_workspace_handler_or_critical_side_channel, try_schedule_next_job, CanceledBy,
    FlowRunners, MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload,
    WrappedError,
};

use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::audit::AuditAuthor;
use windmill_queue::{canceled_job_to_result, push};

lazy_static::lazy_static! {
    /// Per-worker LRU cache of resolved `flow_env` values, keyed by flow job id.
    /// `update_flow_status_after_job_completion_internal` runs once per child-step
    /// completion, and re-resolving `$var:`/`$res:` references each time was the
    /// dominant heap-allocation source under flow-heavy load. The cache aligns
    /// predicate evaluation with `handle_flow`'s input-transform path, which
    /// already resolves once at flow entry — predicates were the only place still
    /// re-resolving on every read.
    ///
    /// Per-worker scope (caveat): the cache lives in process memory, not the DB,
    /// so different workers processing children of the same flow each compute
    /// their own snapshot on first miss. If a `$var:`/`$res:` value mutates
    /// mid-flow, predicate eval on different workers can observe different
    /// values for the same flow run. For typical use (env values configured at
    /// flow start, read-only during execution) this is invisible. Cross-worker
    /// determinism would require persisting the resolved env in `v2_job_status`;
    /// see follow-up notes.
    ///
    /// Entries become dead weight once a flow completes and are evicted by LRU
    /// pressure. Bounded at 1024 entries; per-entry footprint depends on the
    /// env's contents (literals are small but a single resolved `$res:` can be
    /// tens of KB), so worst-case memory scales with workload mix rather than
    /// being fixed.
    static ref RESOLVED_FLOW_ENV_CACHE: quick_cache::sync::Cache<Uuid, Arc<HashMap<String, Box<RawValue>>>> =
        quick_cache::sync::Cache::new(1024);
}

#[derive(Debug)]
pub struct SchedulePushZombieError(pub String);

impl std::fmt::Display for SchedulePushZombieError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl std::error::Error for SchedulePushZombieError {}

/// Helper function to write itered data to separate table
/// Returns None if data was written to separate table, Some(itered) if it should be stored in JSONB
async fn write_itered_to_db(
    db: &DB,
    job_id: Uuid,
    itered: &Vec<Box<RawValue>>,
) -> error::Result<Option<Vec<Box<RawValue>>>> {
    if MIN_VERSION_IS_AT_LEAST_1_595.met().await {
        // Write to separate table
        sqlx::query!(
            "INSERT INTO flow_iterator_data (job_id, itered) VALUES ($1, $2)
             ON CONFLICT (job_id) DO UPDATE SET itered = $2",
            job_id,
            Json(itered) as Json<&Vec<Box<RawValue>>>,
        )
        .execute(db)
        .await?;
        // Return None to indicate itered should not be stored in JSONB
        Ok(None)
    } else {
        // Return Some(itered) to indicate it should be stored in JSONB for backwards compatibility
        Ok(Some(itered.clone()))
    }
}

/// Helper function to read itered data from separate table
/// Falls back to reading from JSONB flow_status if not found in separate table or version too old
async fn read_itered_from_db(
    db: &DB,
    job_id: Uuid,
    itered_from_status: &Option<Vec<Box<RawValue>>>,
) -> error::Result<Vec<Box<RawValue>>> {
    // Only try to read from separate table if version supports it
    if MIN_VERSION_IS_AT_LEAST_1_595.met().await {
        let result = sqlx::query_scalar!(
            "SELECT itered as \"itered: Json<Vec<Box<RawValue>>>\" FROM flow_iterator_data WHERE job_id = $1",
            job_id,
        )
        .fetch_optional(db)
        .await?;

        if let Some(Json(itered)) = result {
            // Found in separate table
            return Ok(itered);
        }
    }

    // Fall back to reading from JSONB flow_status (backwards compatibility or not found in table)
    Ok(itered_from_status.clone().unwrap_or_default()) // can be none for restarted flows, in which case we return an empty vector
}

// #[instrument(level = "trace", skip_all)]
pub async fn update_flow_status_after_job_completion(
    db: &DB,
    client: &AuthedClient,
    flow: uuid::Uuid,
    job_id_for_status: &Uuid,
    w_id: &str,
    success: bool,
    canceled_by: Option<CanceledBy>,
    result: Arc<Box<RawValue>>,
    flow_job_duration: Option<FlowJobDuration>,
    unrecoverable: bool,
    same_worker_tx: &SameWorkerSender,
    worker_dir: &str,
    stop_early_override: Option<bool>,
    worker_name: &str,
    job_completed_tx: JobCompletedSender,
    flow_runners: Option<Arc<FlowRunners>>,
    killpill_rx: &tokio::sync::broadcast::Receiver<()>,
    #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<Option<Arc<MiniPulledJob>>> {
    // this is manual tailrecursion because async_recursion blows up the stack
    potentially_crash_for_testing();

    let mut rec = RecUpdateFlowStatusAfterJobCompletion {
        flow,
        job_id_for_status: job_id_for_status.clone(),
        success,
        canceled_by,
        result,
        flow_job_duration,
        stop_early_override,
        has_triggered_error_handler: false,
    };
    let mut unrecoverable = unrecoverable;
    loop {
        potentially_crash_for_testing();
        let nrec = match Box::pin(update_flow_status_after_job_completion_internal(
            db,
            client,
            rec.flow,
            &rec.job_id_for_status,
            w_id,
            rec.success,
            rec.canceled_by,
            rec.flow_job_duration.clone(),
            rec.result,
            unrecoverable,
            same_worker_tx,
            worker_dir,
            rec.stop_early_override,
            rec.has_triggered_error_handler,
            worker_name,
            job_completed_tx.clone(),
            flow_runners.clone(),
            killpill_rx,
            #[cfg(feature = "benchmark")]
            bench,
        ))
        .await
        {
            Ok(j) => j,
            Err(e) => {
                tracing::error!("Error while updating flow status of {} after  completion of {}, updating flow status again with error: {e:#}", rec.flow, &rec.job_id_for_status);
                Box::pin(update_flow_status_after_job_completion_internal(
                    db,
                    client,
                    rec.flow,
                    &rec.job_id_for_status,
                    w_id,
                    false,
                    None,
                    rec.flow_job_duration,
                    Arc::new(to_raw_value(&Json(&WrappedError {
                        error: json!(e.to_string()),
                    }))),
                    true,
                    same_worker_tx,
                    worker_dir,
                    rec.stop_early_override,
                    rec.has_triggered_error_handler,
                    worker_name,
                    job_completed_tx.clone(),
                    flow_runners.clone(),
                    killpill_rx,
                    #[cfg(feature = "benchmark")]
                    bench,
                ))
                .await?
            }
        };
        unrecoverable = false;

        match nrec {
            UpdateFlowStatusAfterJobCompletion::Done(job) => {
                add_time!(bench, "update flow status internal END");
                return Ok(Some(job));
            }
            UpdateFlowStatusAfterJobCompletion::Rec(nrec) => {
                rec = nrec;
            }
            UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch => {
                add_time!(bench, "update flow status internal END");
                return Ok(None);
            }
            UpdateFlowStatusAfterJobCompletion::NotDone => {
                add_time!(bench, "update flow status internal END");
                return Ok(None);
            }
            UpdateFlowStatusAfterJobCompletion::PreprocessingStep => {
                add_time!(bench, "update flow status preprocessing step END");
                return Ok(None);
            }
        }
    }
}

pub enum UpdateFlowStatusAfterJobCompletion {
    Rec(RecUpdateFlowStatusAfterJobCompletion),
    Done(Arc<MiniPulledJob>),
    NotDone,
    NonLastParallelBranch,
    PreprocessingStep,
}

pub struct RecUpdateFlowStatusAfterJobCompletion {
    flow: uuid::Uuid,
    job_id_for_status: Uuid,
    success: bool,
    result: Arc<Box<RawValue>>,
    canceled_by: Option<CanceledBy>,
    flow_job_duration: Option<FlowJobDuration>,
    stop_early_override: Option<bool>,
    has_triggered_error_handler: bool,
}

#[derive(Deserialize)]
struct RecoveryObject {
    recover: Option<bool>,
}

/// Returns `(skip_if_stopped, error_message, include_step_result)`.
fn get_stop_after_if_data(stop_after_if: Option<&StopAfterIf>) -> (bool, Option<String>, bool) {
    if let Some(stop_after_if) = stop_after_if {
        // skip_if_stopped and error_message are mutually exclusive:
        // skip_if_stopped=true means clean stop (mark remaining as skipped),
        // error_message means stop with error. skip_if_stopped takes precedence.
        if stop_after_if.skip_if_stopped {
            return (true, None, false);
        }
        let err_msg = stop_after_if.error_message.as_ref().and_then(|message| {
            if message.is_empty() {
                Some(format!("stop after if: {}", stop_after_if.expr))
            } else {
                Some(message.clone())
            }
        });
        return (false, err_msg, stop_after_if.error_include_result);
    }
    return (false, None, false);
}

async fn get_id_ctx_for_expr(
    expr: &str,
    flow: uuid::Uuid,
    db: &DB,
    status: &FlowStatus,
) -> error::Result<Option<IdContext>> {
    if expr.contains("results.") || expr.contains("results[") || expr.contains("results?.") {
        let flow_job = get_mini_pulled_job(db, &flow).await?;
        if let Some(flow_job) = flow_job {
            Ok(Some(get_transform_context(&flow_job, "", &status)))
        } else {
            Ok(None)
        }
    } else {
        Ok(None)
    }
}

async fn evaluate_stop_after_all_iters_if(
    db: &DB,
    stop_after_all_iters_if: &StopAfterIf,
    module_status: &FlowStatusModule,
    w_id: &str,
    client: &AuthedClient,
    stop_early: &mut bool,
    skip_if_stop_early: &mut bool,
    stop_early_err_msg: &mut Option<String>,
    stop_early_include_result: &mut bool,
    nresult: &mut Option<Arc<Box<RawValue>>>,
    args: HashMap<String, Box<RawValue>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    flow: uuid::Uuid,
    status: &FlowStatus,
) -> error::Result<()> {
    let iters_result = match &module_status {
        FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), .. } => {
            Arc::new(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?)
        }
        _ => {
            return Err(Error::internal_err(format!(
                "A branchall or loop should have flow_jobs"
            )))
        }
    };

    *nresult = Some(iters_result.clone()); // as an optimization, we store the result of all jobs as when stop_early_after_all_iters evaluates to false, it would have to be computed (finished loop/branchall)

    let id_ctx = get_id_ctx_for_expr(&stop_after_all_iters_if.expr, flow, db, status).await?;

    let stop_early_after_all_iters = compute_bool_from_expr(
        &stop_after_all_iters_if.expr,
        Marc::new(args),
        flow_env,
        iters_result.clone(),
        None,
        id_ctx.as_ref(),
        Some(client),
        None,
        None,
    )
    .await?;

    if stop_early_after_all_iters {
        *stop_early = true;
        (
            *skip_if_stop_early,
            *stop_early_err_msg,
            *stop_early_include_result,
        ) = get_stop_after_if_data(Some(stop_after_all_iters_if));
    }
    Ok(())
}

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,
    client: &AuthedClient,
    flow: uuid::Uuid,
    job_id_for_status: &Uuid,
    w_id: &str,
    mut success: bool,
    canceled_by: Option<CanceledBy>,
    mut flow_job_duration: Option<FlowJobDuration>,
    result: Arc<Box<RawValue>>,
    unrecoverable: bool,
    same_worker_tx: &SameWorkerSender,
    worker_dir: &str,
    stop_early_override: Option<bool>,
    has_triggered_error_handler: bool,
    worker_name: &str,
    job_completed_tx: JobCompletedSender,
    flow_runners: Option<Arc<FlowRunners>>,
    killpill_rx: &tokio::sync::broadcast::Receiver<()>,
    #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<UpdateFlowStatusAfterJobCompletion> {
    let mut has_triggered_error_handler = has_triggered_error_handler;
    add_time!(bench, "update flow status internal START");
    struct ChatAiInfo {
        chat_input_enabled: bool,
        conversation_id: Option<Uuid>,
        is_ai_agent_step: bool,
        omit_output_from_conversation: bool,
    }
    let (
        should_continue_flow,
        flow_job,
        flow_data,
        stop_early,
        skip_if_stop_early,
        nresult,
        is_failure_step,
        _cleanup_module,
        chat_ai_info,
    ) = {
        // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}");

        let (job_kind, script_hash, old_status, raw_flow) = sqlx::query!(
             "SELECT
                 kind AS \"job_kind!: JobKind\",
                 runnable_id AS \"script_hash: ScriptHash\",
                 flow_status AS \"flow_status!: Json<Box<RawValue>>\",
                 raw_flow AS \"raw_flow: Json<Box<RawValue>>\"
             FROM v2_job INNER JOIN v2_job_status ON v2_job.id = v2_job_status.id WHERE v2_job.id = $1 AND v2_job.workspace_id = $2 LIMIT 1",
             flow,
             w_id
         )
         .fetch_one(db)
         .await
         .map_err(|e| {
             Error::internal_err(format!(
                 "fetching flow status {flow} while reporting {success} {result:?}: {e:#}"
             ))
         })
         .and_then(|record| {
             Ok((
                 record.job_kind,
                 record.script_hash,
                 serde_json::from_str::<FlowStatus>(record.flow_status.0.get()).map_err(|e| {
                     Error::internal_err(format!(
                         "requiring current module to be parsable as FlowStatus: {e:?}"
                     ))
                 })?,
                 record.raw_flow,
             ))
         })?;

        let flow_data = cache::job::fetch_flow(db, &job_kind, script_hash)
            .or_else(|_| cache::job::fetch_preview_flow(db, &flow, raw_flow))
            .await?;
        let flow_value = flow_data.value();

        let module_step = Step::from_i32_and_len(old_status.step, old_status.modules.len());
        let current_module = match module_step {
            Step::Step { idx: i, .. } => flow_value.modules.get(i),
            _ => None,
        };

        if current_module.is_some_and(|x| x.is_flow()) {
            has_triggered_error_handler = false;
        }

        // Resolve flow_env for predicate evaluations (stop_after_if,
        // stop_after_all_iters_if, retry_if). Two-level gate:
        //   1. Structural — at least one such predicate is configured.
        //   2. Textual — the predicate expression actually references
        //      `flow_env`. Mirrors the existing `expr.contains("results.")`
        //      check in `get_id_ctx_for_expr`. A substring match has false
        //      positives (harmless — just runs an unneeded resolve) but no
        //      false negatives, since `flow_env` must appear textually for the
        //      expression engine to read it.
        let expr_uses_flow_env = |expr: &str| expr.contains("flow_env");
        let retry_if_uses_flow_env = |module: &FlowModule| {
            module
                .retry
                .as_ref()
                .and_then(|r| r.retry_if.as_ref())
                .is_some_and(|r| expr_uses_flow_env(&r.expr))
        };
        let needs_flow_env = current_module.is_some_and(|m| {
            m.stop_after_if
                .as_ref()
                .is_some_and(|s| expr_uses_flow_env(&s.expr))
                || m.stop_after_all_iters_if
                    .as_ref()
                    .is_some_and(|s| expr_uses_flow_env(&s.expr))
                || retry_if_uses_flow_env(m)
        }) || flow_value
            .failure_module
            .as_ref()
            .is_some_and(|fm| retry_if_uses_flow_env(fm));
        // The resolved env is constant for a flow's lifetime, so cache it by flow
        // job id and reuse across child-step completions. Cache miss falls back to
        // the existing resolve+persist path; same-worker subsequent completions are
        // a single Arc::clone away. Transient-failure fallbacks (`is_cacheable ==
        // false`) bypass the insert so a single API blip doesn't poison the rest
        // of the flow run.
        let resolved_flow_env: Option<Arc<HashMap<String, Box<RawValue>>>> = if needs_flow_env {
            if let Some(cached) = RESOLVED_FLOW_ENV_CACHE.get(&flow) {
                Some(cached)
            } else {
                resolve_flow_env_for_status_update(db, client, flow, w_id, flow_value)
                    .await
                    .map(|(env, is_cacheable)| {
                        let arc = Arc::new(env);
                        if is_cacheable {
                            RESOLVED_FLOW_ENV_CACHE.insert(flow, arc.clone());
                        }
                        arc
                    })
            }
        } else {
            None
        };

        let module_status = match module_step {
            Step::PreprocessorStep => old_status
                .preprocessor_module
                .as_ref()
                .ok_or_else(|| Error::internal_err(format!("preprocessor module not found")))?,
            Step::FailureStep => &old_status.failure_module.module_status,
            Step::Step { idx: i, .. } => old_status
                .modules
                .get(i as usize)
                .ok_or_else(|| Error::internal_err(format!("module {i} not found")))?,
        };

        // tracing::debug!(
        //     "UPDATE FLOW STATUS 2: {module_step:#?} {module_status:#?} {old_status:#?} "
        // );

        let (is_loop, skip_loop_failures, parallelism, parallel_loop) =
            if let FlowStatusModule::InProgress { iterator: Some(_), parallel, .. } = module_status
            {
                let value = current_module
                    .as_ref()
                    .and_then(|x| x.get_value_with_skip_failures().ok());
                (
                    true,
                    value
                        .as_ref()
                        .and_then(|x| x.skip_failures)
                        .unwrap_or(false),
                    value.and_then(|x| x.parallelism),
                    *parallel,
                )
            } else {
                (false, false, None, false)
            };

        let (is_branch_all, parallel_branchall) = match module_status {
            FlowStatusModule::InProgress { branchall: Some(_), parallel, .. } => (true, *parallel),
            _ => (false, false),
        };

        // 0 length flows are not failure steps
        let is_failure_step =
            old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;

        let is_flow_stop_early_override = stop_early_override.is_some() && {
            let step = module_step.get_step_index();

            if let Some(_) = step {
                #[derive(Deserialize)]
                struct GetType<'j> {
                    r#type: &'j str,
                }

                current_module
                    .map(|module| {
                        serde_json::from_str::<GetType>(module.value.get())
                            .map(|v| v.r#type == "flow")
                    })
                    .unwrap_or(Ok(false))
                    .unwrap_or(false)
            } else {
                false
            }
        };

        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;
            let args =
                args.map(|flow_args| flow_args.map(|flow_args| flow_args.0).unwrap_or_default());

            args
        }));

        let from_result_to_args = |args: &Result<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())
        };

        // Pre-compute whether the completed job was an identity (skipped) job.
        // When a step has skip_if set and the condition was true, the step runs
        // as an identity job.  We must skip the stop_after_if evaluation in that
        // case because the result is just a pass-through of the previous step's
        // result, not the output of the actual step logic.
        let is_identity_job = if current_module.is_some_and(|m| m.skip_if.is_some()) {
            sqlx::query_scalar!(
                "SELECT kind = 'identity' FROM v2_job WHERE id = $1",
                job_id_for_status
            )
            .fetch_optional(db)
            .await
            .map_err(|e| Error::internal_err(format!("error during identity job check: {e:#}")))?
            .flatten()
            .unwrap_or(false)
        } else {
            false
        };

        let (
            mut stop_early,
            mut stop_early_err_msg,
            mut skip_if_stop_early,
            mut stop_early_include_result,
            continue_on_error,
        ) = if stop_early_override.is_some()
            && !is_flow_stop_early_override
            && !parallel_loop
            && !parallel_branchall
        {
            // we ignore stop_early_override (stop_early in children) if module is parallel or is a flow step
            let se = stop_early_override.as_ref().unwrap();
            (true, None, *se, false, false)
        } else if is_failure_step || module_step.is_preprocessor_step() {
            (false, None, false, false, false)
        } else if let Some(current_module) = current_module {
            let stop_early = success
                    && !is_branch_all // we don't support stop_early per branch
                    && !parallel_loop // we don't support anymore stop_early per iteration when parallel for loop (removed from frontend)
                    && !is_identity_job // don't evaluate stop_after_if for skipped (identity) steps
                    && if let Some(expr) = current_module
                        .stop_after_if
                        .as_ref()
                        .map(|x| x.expr.as_str())
                    {
                        let all_iters =
                            match &module_status {
                                FlowStatusModule::InProgress {
                                    flow_jobs: Some(flow_jobs), ..
                                } if expr.contains("all_iters") => Some(Arc::new(
                                    retrieve_flow_jobs_results(db, w_id, flow_jobs).await?,
                                )),
                                _ => None,
                            };
                        let args = from_result_to_args(args.as_ref().await.get_ref())?;

                        let id_ctx = get_id_ctx_for_expr(expr, flow, db, &old_status).await?;

                        let bool_res = compute_bool_from_expr(
                            &expr,
                            Marc::new(args),
                            resolved_flow_env.as_deref(),
                            result.clone(),
                            all_iters,
                            id_ctx.as_ref(),
                            Some(client),
                            None,
                            None,
                        )
                        .await;
                        if let Err(e) = &bool_res {
                            append_predicate_error_to_root_logs(
                                db,
                                fetch_root_flow_id(db, flow).await,
                                w_id,
                                Some(&current_module.id),
                                "stop_after_if",
                                expr,
                                e,
                            )
                            .await;
                        }
                        bool_res?
                    } else {
                        false
                    };
            let (skip_if_stopped, stop_early_err_msg, include_result) = if stop_early {
                get_stop_after_if_data(current_module.stop_after_if.as_ref())
            } else {
                (false, None, false)
            };

            (
                stop_early,
                stop_early_err_msg,
                skip_if_stopped,
                include_result,
                current_module.continue_on_error.unwrap_or(false),
            )
        } else {
            (false, None, false, false, false)
        };

        let skip_seq_branch_failure = match module_status {
            FlowStatusModule::InProgress {
                branchall: Some(BranchAllStatus { branch, .. }),
                parallel: false,
                ..
            } => {
                compute_skip_branchall_failure(branch.to_owned(), false, current_module, &None)
                    .await?
            }
            _ => false,
        };

        let mut tx = db.begin().await?;

        add_time!(bench, "process module status START");

        let mut nresult = None;
        let (inc_step_counter, new_status) = match module_status {
            FlowStatusModule::InProgress {
                iterator,
                branchall,
                parallel,
                flow_jobs: Some(jobs),
                flow_jobs_success,
                flow_jobs_duration,
                ..
            } if *parallel => {
                let (nindex, len) = match (iterator, branchall) {
                    (Some(FlowIterator { itered, itered_len, .. }), _) => {
                        let position = if flow_jobs_success.is_some() {
                            find_flow_job_index(jobs, job_id_for_status)
                        } else {
                            None
                        };

                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                                 "
                                 UPDATE v2_job_status SET flow_status = 
                                    JSONB_SET(JSONB_SET(JSONB_SET(JSONB_SET(
                                            flow_status, 
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), 
                                            ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb),
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5),
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6)
                                 WHERE id = $2
                                 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
                                 old_status.step,
                                 flow,
                                 position as i32,
                                 json!(success),
                                 flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default(),
                                 flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(),
                             )
                         } else {
                             sqlx::query_scalar!(
                                 "UPDATE v2_job_status SET
                                     flow_status = JSONB_SET(
                                         flow_status,
                                         ARRAY['modules', $1::TEXT, 'iterator', 'index'],
                                         ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb
                                     )
                                 WHERE id = $2
                                 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
                                 old_status.step,
                                 flow
                             )
                         }
                         .fetch_one(&mut *tx)
                         .await.map_err(|e| {
                             Error::internal_err(format!(
                                 "error while fetching iterator index: {e:#}"
                             ))
                         })?.ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress for flow {flow} at step {}", old_status.step)))?;

                        // let status_for_debug = sqlx::query!(
                        //     "SELECT flow_status FROM v2_job_status WHERE id = $1",
                        //     flow
                        // )
                        // .fetch_one(&mut *tx)
                        // .await
                        // .map_err(|e| {
                        //     Error::internal_err(format!("error while fetching flow status: {e:#}"))
                        // })?;

                        // tracing::error!("status_for_debug: {:?}", status_for_debug.flow_status);

                        let itered_len = if let Some(itered_len) = itered_len {
                            *itered_len
                        } else {
                            itered.as_ref().map(|itered| itered.len()).unwrap_or(0)
                        };

                        tracing::info!(
                             "parallel iteration {job_id_for_status} of flow {flow} update nindex: {nindex} len: {len}",
                             nindex = nindex,
                             len = itered_len
                         );
                        (nindex, itered_len as i32)
                    }
                    (_, Some(BranchAllStatus { len, .. })) => {
                        let position = if flow_jobs_success.is_some() {
                            find_flow_job_index(jobs, job_id_for_status)
                        } else {
                            None
                        };

                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                                 "UPDATE v2_job_status SET flow_status = 
                                 CASE 
                                 WHEN flow_status->'modules'->$1::int->'flow_jobs_duration' IS NOT NULL THEN
                                    JSONB_SET(
                                         JSONB_SET(JSONB_SET(JSONB_SET(
                                            flow_status, 
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), 
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), 
                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),
                                            ARRAY['modules', $1::TEXT, 'branchall', 'branch'],
                                            ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb
                                     )
                                 ELSE
                                    JSONB_SET(JSONB_SET(
                                    flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),
                                         ARRAY['modules', $1::TEXT, 'branchall', 'branch'],
                                         ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb
                                     )
                                 END
                                 WHERE id = $2
                                 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
                                 old_status.step,
                                 flow,
                                 position as i32,
                                 json!(success),
                                 flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default(),
                                 flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(),
                             )
                         } else {
                             sqlx::query_scalar!(
                                 "UPDATE v2_job_status SET
                                     flow_status = JSONB_SET(
                                         flow_status,
                                         ARRAY['modules', $1::TEXT, 'branchall', 'branch'],
                                         ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb
                                     )
                                 WHERE id = $2
                                 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
                                 old_status.step,
                                 flow
                             )
                         }
                         .fetch_one(&mut *tx)
                         .await
                         .map_err(|e| {
                             Error::internal_err(format!(
                                 "error while fetching branchall index: {e:#}"
                             ))
                         })?
                         .ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress")))?;
                        (nindex, *len as i32)
                    }
                    _ => Err(Error::internal_err(format!(
                        "unexpected status for parallel module"
                    )))?,
                };
                // reset ping after updating flow status:
                let _ = sqlx::query!("UPDATE v2_job_runtime SET ping = NULL WHERE id = $1", flow)
                    .execute(&mut *tx)
                    .await?;
                if nindex == len {
                    let success_and_durations =  match sqlx::query!(
                        "SELECT flow_status->'modules'->$2::int->'flow_jobs_success' as \"flow_jobs_success: Json<Vec<Option<bool>>>\", flow_status->'modules'->$2::int->'flow_jobs_duration' as \"flow_jobs_duration: Json<FlowJobsDuration>\"
                        FROM v2_job_status WHERE id = $1",
                        flow,
                        old_status.step
                    )
                    .fetch_one(&mut *tx)
                    .await {
                        Err(e) => {
                            tracing::error!("error while fetching success and durations: {e:#}");
                            (flow_jobs_success.clone(), flow_jobs_duration.clone())
                        }
                        Ok(x) => (x.flow_jobs_success.map(|x| x.0), x.flow_jobs_duration.map(|x| x.0)),
                    };

                    let mut flow_jobs_success = success_and_durations.0;
                    if let Some(flow_job_success) = flow_jobs_success.as_mut() {
                        let position = jobs.iter().position(|x| x == job_id_for_status);
                        if let Some(position) = position {
                            if position < flow_job_success.len() {
                                flow_job_success[position] = Some(success);
                            }
                        }
                    }
                    let mut flow_jobs_duration = success_and_durations.1;
                    if let Some(flow_jobs_duration) = flow_jobs_duration.as_mut() {
                        let position = jobs.iter().position(|x| x == job_id_for_status);
                        flow_jobs_duration.set(position, &flow_job_duration);
                    }

                    let branches = current_module
                        .and_then(|x| x.get_branches_skip_failures().ok())
                        .map(|x| {
                            x.branches
                                .iter()
                                .map(|b| b.skip_failure.unwrap_or(false))
                                .collect::<Vec<_>>()
                        });

                    let mut njobs = Vec::new();

                    let jobs_filtered = if branchall.is_some() {
                        if let Some(branches) = branches {
                            for (branch, job) in branches.iter().zip(jobs.iter()) {
                                if !branch {
                                    njobs.push(job.clone());
                                }
                            }
                            njobs.as_slice()
                        } else {
                            jobs.as_slice()
                        }
                    } else {
                        jobs.as_slice()
                    };

                    // evaluate stop_after_all_iters_if for parallel loops/branchall
                    if let Some(stop_after_all_iters_if) = current_module
                        .as_ref()
                        .and_then(|x| x.stop_after_all_iters_if.as_ref())
                    {
                        let args = from_result_to_args(args.as_ref().await.get_ref())?;
                        if let Err(e) = evaluate_stop_after_all_iters_if(
                            db,
                            stop_after_all_iters_if,
                            module_status,
                            w_id,
                            client,
                            &mut stop_early,
                            &mut skip_if_stop_early,
                            &mut stop_early_err_msg,
                            &mut stop_early_include_result,
                            &mut nresult,
                            args,
                            resolved_flow_env.as_deref(),
                            flow,
                            &old_status,
                        )
                        .await
                        {
                            tracing::error!("error evaluating stop_after_all_iters_if: {e:#}");
                            stop_early = true;
                            skip_if_stop_early = false;
                            stop_early_err_msg = Some(format!(
                                "Error evaluating stop_after_all_iters_if expression `{}`: {e:#}",
                                stop_after_all_iters_if.expr
                            ));
                        }
                    }

                    let new_status = if
                        !(stop_early && stop_early_err_msg.is_some() && !skip_if_stop_early) // if stop_early with error and NOT skip_if_stopped, mark as failure
                        && (
                                skip_loop_failures
                                || sqlx::query_scalar!(
                                    "SELECT status = 'success' OR status = 'skipped' AS \"success!\" FROM v2_job_completed WHERE id = ANY($1)",
                                    jobs_filtered
                                )
                                .fetch_all(&mut *tx)
                                .await
                                .map_err(|e| {
                                    Error::internal_err(format!(
                                        "error while fetching success from completed_jobs: {e:#}"
                                    ))
                                })?
                                .into_iter()
                                .all(|x| x)
                            )
                     {
                         success = true;
                         FlowStatusModule::Success {
                             id: module_status.id(),
                             job: job_id_for_status.clone(),
                             flow_jobs: Some(jobs.clone()),
                             flow_jobs_success: flow_jobs_success.clone(),
                             flow_jobs_duration: flow_jobs_duration.clone(),
                             branch_chosen: None,
                             approvers: vec![],
                             failed_retries: vec![],
                             skipped: stop_early && skip_if_stop_early,
                             agent_actions: None,
                             agent_actions_success: None,
                         }
                     } else {
                         success = false;
                         FlowStatusModule::Failure {
                             id: module_status.id(),
                             job: job_id_for_status.clone(),
                             flow_jobs: Some(jobs.clone()),
                             flow_jobs_success: flow_jobs_success.clone(),
                             flow_jobs_duration: flow_jobs_duration.clone(),
                             branch_chosen: None,
                             failed_retries: vec![],
                             agent_actions: None,
                             agent_actions_success: None,
                         }
                     };
                    let r = sqlx::query_scalar!(
                         "DELETE FROM parallel_monitor_lock WHERE parent_flow_id = $1 RETURNING last_ping",
                         flow,
                     ).fetch_optional(db).await.map_err(|e| {
                         Error::internal_err(format!(
                             "error while deleting parallel_monitor_lock: {e:#}"
                         ))
                     })?;

                    if r.is_some() {
                        tracing::info!(
                            "parallel flow has removed lock on its parent, last ping was {:?}",
                            r.unwrap()
                        );
                    }
                    tracing::info!(
                        "parallel iteration {job_id_for_status} of flow {flow} has finished",
                    );

                    // for parallel branchall and forloop, we do not want to trigger the error handler again at the forloop/branchall node since it was already triggered at the leaf level
                    // so we want to ignore the has_triggered_error_handler flag at the forloop/branchall node and reset it based on if the node is a success or failure
                    if !success && flow_value.failure_module.is_some() {
                        has_triggered_error_handler = true;
                    } else {
                        has_triggered_error_handler = false;
                    }
                    (success, Some(new_status))
                } else {
                    add_time!(bench, "handle parallel flow start");
                    tx.commit().await?;

                    if parallelism.is_some() {
                        sqlx::query!(
                            "UPDATE v2_job_queue q SET suspend = 0
                             FROM v2_job j, v2_job_status f
                             WHERE q.workspace_id = $1 AND q.suspend = $3 AND j.parent_job = $2
                                 AND f.id = j.id AND q.id = j.id
                                 AND (f.flow_status->'step')::int = 0",
                            w_id,
                            flow,
                            nindex
                        )
                        .execute(db)
                        .await
                        .map_err(|e| {
                            Error::internal_err(format!(
                                "error resuming job at suspend {nindex} and parent {flow}: {e:#}"
                            ))
                        })?;
                    }

                    let r = sqlx::query_scalar!(
                         "DELETE FROM parallel_monitor_lock WHERE parent_flow_id = $1 and job_id = $2 RETURNING last_ping",
                         flow,
                         job_id_for_status
                     ).fetch_optional(db).await.map_err(|e| {
                         Error::internal_err(format!("error while removing parallel_monitor_lock: {e:#}"))
                     })?;
                    if r.is_some() {
                        tracing::info!(
                            "parallel flow has removed lock on its parent, last ping was {:?}",
                            r.unwrap()
                        );
                    }
                    add_time!(bench, "non final parallel flow finished");
                    return Ok(UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch);
                }
            }
            FlowStatusModule::InProgress {
                iterator: Some(FlowIterator { index, itered_len, itered }),
                flow_jobs_success,
                flow_jobs,
                while_loop,
                ..
            } if {
                let itered_len = if let Some(itered_len) = itered_len {
                    *itered_len
                } else {
                    // backwards compatibility
                    itered.as_ref().map(|itered| itered.len()).unwrap_or(0)
                };
                (*while_loop || *index + 1 < itered_len)
                    && (success || skip_loop_failures)
                    && !stop_early
            } =>
            {
                if let Some(jobs) = flow_jobs {
                    set_success_and_duration_in_flow_job_success(
                        flow_jobs_success,
                        jobs,
                        job_id_for_status,
                        old_status.step,
                        flow,
                        success,
                        flow_job_duration.clone(),
                        &mut tx,
                    )
                    .await?;
                }

                (false, None)
            }
            FlowStatusModule::InProgress {
                branchall: Some(BranchAllStatus { branch, len, .. }),
                flow_jobs_success,
                flow_jobs,
                ..
            } if branch.to_owned() < len - 1
                && (success || skip_seq_branch_failure)
                && !stop_early =>
            {
                if let Some(jobs) = flow_jobs {
                    set_success_and_duration_in_flow_job_success(
                        flow_jobs_success,
                        jobs,
                        job_id_for_status,
                        old_status.step,
                        flow,
                        success,
                        flow_job_duration.clone(),
                        &mut tx,
                    )
                    .await?;
                }
                (false, None)
            }
            _ => {
                // this case is when when not a parallel loops/branchall and not an in progress loop/branchall

                if stop_early && is_loop {
                    // if we're stopping early inside a (non-parallel) loop, we don't want to bubble up the stop_early to the parent => we only want to break the loop (see conditions in match above)
                    stop_early = false;
                    stop_early_err_msg = None;
                    skip_if_stop_early = false;
                    stop_early_include_result = false;
                }

                if is_loop || (is_branch_all && !stop_early) {
                    // when we finish a loop or branchall, we only want to evaluate stop_after_all_iters_if if:
                    //  -  we're in a loop (non-parallel)
                    //  -  we're in a branchall and it wasn't stopped early from inside (non-parallel branchall stopped inside => stop flow => no need to evaluate stop_after_all_iters_if)
                    if let Some(stop_after_all_iters_if) = current_module
                        .as_ref()
                        .and_then(|x| x.stop_after_all_iters_if.as_ref())
                    {
                        let args = from_result_to_args(args.as_ref().await.get_ref())?;

                        if let Err(e) = evaluate_stop_after_all_iters_if(
                            db,
                            stop_after_all_iters_if,
                            module_status,
                            w_id,
                            client,
                            &mut stop_early,
                            &mut skip_if_stop_early,
                            &mut stop_early_err_msg,
                            &mut stop_early_include_result,
                            &mut nresult,
                            args,
                            resolved_flow_env.as_deref(),
                            flow,
                            &old_status,
                        )
                        .await
                        {
                            stop_early = true;
                            skip_if_stop_early = false;
                            stop_early_err_msg = Some(format!(
                                "Error evaluating stop_after_all_iters_if expression `{}`: {e:#}",
                                stop_after_all_iters_if.expr
                            ));
                        }
                    }
                }

                let flow_jobs = module_status.flow_jobs();
                let branch_chosen = module_status.branch_chosen();
                let mut flow_jobs_success = module_status.flow_jobs_success();
                let mut flow_jobs_duration = module_status.flow_jobs_duration();

                if let (Some(flow_job_success), Some(flow_jobs)) =
                    (flow_jobs_success.as_mut(), flow_jobs.as_ref())
                {
                    let position = flow_jobs.iter().position(|x| x == job_id_for_status);
                    if let Some(position) = position {
                        if position < flow_job_success.len() {
                            flow_job_success[position] = Some(success);
                        }
                    }
                }

                if let (Some(flow_jobs_duration), Some(flow_jobs)) =
                    (flow_jobs_duration.as_mut(), flow_jobs.as_ref())
                {
                    let position = flow_jobs.iter().position(|x| x == job_id_for_status);
                    flow_jobs_duration.set(position, &flow_job_duration);
                }

                // if stop_early with error message and NOT skip_if_stopped, mark as failure
                // if skip_if_stopped=true, we want to mark as success (skipped), not failure
                if (success
                    || (flow_jobs.is_some() && (skip_loop_failures || skip_seq_branch_failure)))
                    && !(stop_early && stop_early_err_msg.is_some() && !skip_if_stop_early)
                {
                    let is_skipped = (stop_early && skip_if_stop_early)
                        || if current_module.as_ref().is_some_and(|m| m.skip_if.is_some()) {
                            is_identity_job
                        } else {
                            false
                        };
                    success = true;
                    (
                        true,
                        Some(FlowStatusModule::Success {
                            id: module_status.id(),
                            job: job_id_for_status.clone(),
                            flow_jobs,
                            flow_jobs_success,
                            flow_jobs_duration,
                            branch_chosen,
                            approvers: vec![],
                            failed_retries: old_status.retry.failed_jobs.clone(),
                            skipped: is_skipped,
                            agent_actions: module_status.agent_actions(),
                            agent_actions_success: module_status.agent_actions_success(),
                        }),
                    )
                } else {
                    let inc = if continue_on_error {
                        let retry = current_module
                            .as_ref()
                            .and_then(|x| x.retry.clone())
                            .unwrap_or_default();

                        tracing::info!("update flow status on retry: {retry:#?} ");
                        let args = from_result_to_args(args.as_ref().await.get_ref())?;

                        evaluate_retry(
                            &retry,
                            &old_status.retry,
                            result.clone(),
                            Marc::new(args),
                            resolved_flow_env.as_deref(),
                            Some(client),
                        )
                        .await?
                        .is_none()
                    } else {
                        false
                    };
                    if stop_early && stop_early_err_msg.is_some() {
                        // if stop_early with error message, we need to set explictely success to false as this function was called with success = true as the job succeeded
                        success = false;
                    }
                    (
                        inc,
                        Some(FlowStatusModule::Failure {
                            id: module_status.id(),
                            job: job_id_for_status.clone(),
                            flow_jobs,
                            flow_jobs_success,
                            flow_jobs_duration,
                            branch_chosen,
                            failed_retries: old_status.retry.failed_jobs.clone(),
                            agent_actions: module_status.agent_actions(),
                            agent_actions_success: module_status.agent_actions_success(),
                        }),
                    )
                }
            }
        };

        if stop_early && stop_early_err_msg.is_some() {
            let mut error = serde_json::json!({
                "name": "EarlyStopError",
                "message": stop_early_err_msg.as_ref().unwrap(),
            });
            if stop_early_include_result {
                // Embed the stopping step's own result inside the error object instead
                // of discarding it, keeping the top-level result shape `{ "error": .. }`
                // unchanged. `nresult` is already set for loops/branchall (aggregated
                // iteration results), otherwise fall back to the step result.
                let step_result = nresult.clone().unwrap_or_else(|| result.clone());
                error["result"] =
                    serde_json::to_value(&step_result).unwrap_or(serde_json::Value::Null);
            }
            nresult = Some(Arc::new(to_raw_value(
                &serde_json::json!({ "error": error }),
            )));
        }

        let step_counter = if inc_step_counter {
            sqlx::query!(
                "UPDATE v2_job_status
                 SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)
                 WHERE id = $2",
                json!(old_status.step + 1),
                flow
            )
            .execute(&mut *tx)
            .await
            .map_err(|e| {
                Error::internal_err(format!("error while setting flow index for {flow}: {e:#}"))
            })?;
            old_status.step + 1
        } else {
            old_status.step
        };

        // tracing::error!(
        //     "step_counter: {:?} {} {inc_step_counter} {flow}",
        //     step_counter,
        //     old_status.step,
        // );
        // panic!("stop");

        /* is_last_step is true when the step_counter (the next step index) is an invalid index */
        let is_last_step = usize::try_from(step_counter)
            .map(|i| !(..old_status.modules.len()).contains(&i))
            .unwrap_or(true);

        if let Some(new_status) = new_status.as_ref() {
            if is_failure_step {
                let parent_module = sqlx::query_scalar!(
                     "SELECT flow_status->'failure_module'->>'parent_module' FROM v2_job_status WHERE id = $1",
                     flow
                 )
                 .fetch_one(&mut *tx)
                 .await.map_err(|e| {
                     Error::internal_err(format!(
                         "error while fetching failure module: {e:#}"
                     ))
                 })?;

                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1)
                     WHERE id = $2",
                    json!(FlowStatusModuleWParent {
                        parent_module,
                        module_status: new_status.clone()
                    }),
                    flow
                )
                .execute(&mut *tx)
                .await
                .map_err(|e| {
                    Error::internal_err(format!(
                        "error while setting flow status in failure step: {e:#}"
                    ))
                })?;
            } else if module_step.is_preprocessor_step() {
                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['preprocessor_module'], $1)
                     WHERE id = $2",
                    json!(new_status),
                    flow
                )
                .execute(&mut *tx)
                .await
                .map_err(|e| {
                    Error::internal_err(format!(
                        "error while setting flow status in preprocessing step: {e:#}"
                    ))
                })?;
            } else {
                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)
                     WHERE id = $3",
                    old_status.step.to_string(),
                    json!(new_status),
                    flow
                )
                .execute(&mut *tx)
                .await
                .map_err(|e| {
                    Error::internal_err(format!("error while setting new flow status: {e:#}"))
                })?;

                if let Some(job_result) = new_status.job_result() {
                    sqlx::query!(
                         "UPDATE v2_job_status
                         SET flow_leaf_jobs = JSONB_SET(coalesce(flow_leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)
                         WHERE COALESCE((SELECT flow_innermost_root_job FROM v2_job WHERE id = $3), $3) = id",
                         new_status.id(),
                         json!(job_result),
                         flow
                     )
                     .execute(&mut *tx)
                     .await.map_err(|e| {
                         Error::internal_err(format!(
                             "error while setting leaf jobs: {e:#}"
                         ))
                     })?;
                }
            }
        }

        let nresult = if let Some(nresult) = nresult {
            // can be some either with early stop error or with the flow jobs results (was fetched to evaluate stop_early_after_all_iters but evaluated to false)
            nresult
        } else {
            match &new_status {
                Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. })
                | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => {
                    Arc::new(retrieve_flow_jobs_results(db, w_id, jobs).await?)
                }
                _ => result.clone(),
            }
        };

        if old_status.retry.fail_count > 0
            && matches!(&new_status, Some(FlowStatusModule::Success { .. }))
        {
            sqlx::query!(
                "UPDATE v2_job_status
                 SET flow_status = flow_status - 'retry'
                 WHERE id = $1",
                flow
            )
            .execute(&mut *tx)
            .await
            .context("remove flow status retry")?;
        }

        let flow_job = get_mini_pulled_job(&mut *tx, &flow)
            .await?
            .ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?;
        tx.commit().await?;

        if module_step.is_preprocessor_step() && success {
            let tag_and_concurrency_key = get_tag_and_concurrency(&flow, db).await;
            let has_debouncing = flow_value
                .debouncing_settings
                .debounce_delay_s
                .filter(|x| *x > 0)
                .is_some();
            let concurrency_requires_args = tag_and_concurrency_key.as_ref().is_some_and(|x| {
                x.tag.as_ref().is_some_and(|t| t.contains("$args"))
                    || x.concurrency_key
                        .as_ref()
                        .is_some_and(|ck| ck.contains("$args"))
            });
            let require_args = concurrency_requires_args || has_debouncing;
            let mut tag = tag_and_concurrency_key.as_ref().and_then(|x| x.tag.clone());
            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);
            let concurrency_time_window_s = tag_and_concurrency_key
                .as_ref()
                .and_then(|x| x.concurrency_time_window_s);

            let fetched_args = if require_args {
                let args = sqlx::query_scalar!(
                    "SELECT result  as \"result: Json<HashMap<String, Box<RawValue>>>\"
                 FROM v2_job_completed
                 WHERE id = $1",
                    job_id_for_status
                )
                .fetch_one(db)
                .await
                .map_err(|e| {
                    Error::internal_err(format!("error while fetching preprocessing args: {e:#}"))
                })?;
                Some(args.unwrap_or_default().0)
            } else {
                None
            };

            if concurrency_requires_args {
                let args = PushArgs::from(fetched_args.as_ref().unwrap());
                if let Some(ck) = concurrency_key {
                    insert_concurrency_key(
                        &flow_job.workspace_id,
                        &args,
                        &flow_job.runnable_path,
                        JobKind::Flow,
                        Some(ck),
                        db,
                        flow,
                    )
                    .await?;
                }
                if let Some(t) = tag {
                    tag = Some(interpolate_args(t, &args, &flow_job.workspace_id));
                }
            } else if concurrent_limit.is_some() {
                insert_concurrency_key(
                    &flow_job.workspace_id,
                    &PushArgs::from(&HashMap::new()),
                    &flow_job.runnable_path,
                    JobKind::Flow,
                    concurrency_key,
                    db,
                    flow,
                )
                .await?;
            }

            let scheduled_for: Option<chrono::DateTime<chrono::Utc>> = {
                #[cfg(feature = "private")]
                {
                    if has_debouncing {
                        let empty_hm = HashMap::new();
                        let args = PushArgs::from(fetched_args.as_ref().unwrap_or(&empty_hm));
                        windmill_queue::jobs_ee::maybe_debounce_post_preprocessing(
                            &flow_value.debouncing_settings,
                            &flow_job.runnable_path,
                            &flow_job.workspace_id,
                            flow,
                            &args,
                            db,
                        )
                        .await?
                    } else {
                        None
                    }
                }
                #[cfg(not(feature = "private"))]
                {
                    None
                }
            };

            // Store the flow's debouncing settings in the runnable_settings_handle
            // so that maybe_apply_debouncing can find them after pull and perform
            // argument accumulation. This is needed both when debounced (CanDebounce)
            // and when firing immediately (MaxCountExceeded), since accumulation
            // happens at pull time in both cases.
            let new_runnable_settings_handle: Option<i64> = if has_debouncing {
                let debouncing_hash = flow_value
                    .debouncing_settings
                    .insert_cached(db)
                    .await
                    .unwrap_or_else(|e| {
                        tracing::error!(
                            "Failed to insert debouncing settings for post-preprocessing: {e:#}"
                        );
                        None
                    });
                windmill_common::runnable_settings::insert_rs(
                    windmill_common::runnable_settings::RunnableSettings {
                        debouncing_settings: debouncing_hash,
                        concurrency_settings: None,
                    },
                    db,
                )
                .await
                .unwrap_or_else(|e| {
                    tracing::error!(
                        "Failed to insert runnable settings for post-preprocessing: {e:#}"
                    );
                    None
                })
            } else {
                None
            };

            sqlx::query!(
                "WITH job_result AS (
                 SELECT result
                 FROM v2_job_completed
                 WHERE id = $1
             ),
             updated_queue AS (
                UPDATE v2_job_queue
                SET running = false,
                tag = COALESCE($3, tag),
                scheduled_for = COALESCE($6, scheduled_for),
                runnable_settings_handle = COALESCE($7, runnable_settings_handle)
                WHERE id = $2
             )
             UPDATE v2_job
             SET
                tag = COALESCE($3, tag),
                concurrent_limit = COALESCE($4, concurrent_limit),
                concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),
                args = COALESCE(
                     CASE
                         WHEN job_result.result IS NULL THEN NULL
                         WHEN jsonb_typeof(job_result.result) = 'object'
                         THEN job_result.result
                         WHEN jsonb_typeof(job_result.result) = 'null'
                         THEN NULL
                         ELSE jsonb_build_object('value', job_result.result)
                     END,
                     '{}'::jsonb
                 ),
                 preprocessed = TRUE
             FROM job_result
             WHERE v2_job.id = $2;
             ",
                job_id_for_status,
                flow,
                tag,
                concurrent_limit,
                concurrency_time_window_s,
                scheduled_for,
                new_runnable_settings_handle,
            )
            .execute(db)
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "error while updating args in preprocessing step: {e:#}"
                ))
            })?;
            return Ok(UpdateFlowStatusAfterJobCompletion::PreprocessingStep);
        }

        let job_root = flow_job
            .flow_innermost_root_job
            .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),
                resolved_flow_env.as_deref(),
                Some(client),
            )
            .await?
            .is_some();

            Ok(should_retry)
        };

        // For a regular module with continue_on_error at the last position, nothing else
        // overrides `success` — `flow_jobs` is None so the loop/branchall override above
        // doesn't fire, and `should_continue_flow` resolves to `!is_last_step = false`,
        // letting the flow complete with success=false and bubble the error up to the
        // enclosing job/subflow. Detect that case and treat the flow as successful.
        let recoverable_failure_at_last_step = !success
            && is_last_step
            && !unrecoverable
            && (skip_seq_branch_failure || skip_loop_failures || continue_on_error);

        let should_continue_flow = match success {
            _ if stop_early => stop_early_err_msg.is_some() && flow_value.failure_module.is_some(), // if stop_early_err_msg some, we want to trigger the error handler before stopping the flow, if any
            _ if flow_job.is_canceled() => false,
            true => !is_last_step,
            false if unrecoverable => false,
            false if skip_seq_branch_failure || skip_loop_failures || continue_on_error => {
                !is_last_step
            }
            false if should_retry().await? => true,
            false
                if !is_failure_step
                    && !has_triggered_error_handler
                    && flow_value.failure_module.is_some() =>
            {
                true
            }
            false => false,
        };

        if recoverable_failure_at_last_step {
            success = true;
        }

        tracing::info!(id = %flow_job.id, root_id = %job_root, success = %success, stop_early = %stop_early, is_last_step = %is_last_step, unrecoverable = %unrecoverable,
             skip_seq_branch_failure = %skip_seq_branch_failure, skip_loop_failures = %skip_loop_failures,
             current_module_id = %current_module.map(|x| x.id.clone()).unwrap_or_default(),
            continue_on_error = %continue_on_error, should_continue_flow = %should_continue_flow, "computed if flow should continue");

        let is_ai_agent_step = current_module.is_some_and(|m| m.is_ai_agent());
        let omit_output_from_conversation = match (is_ai_agent_step, current_module) {
            (true, Some(module)) => match module.get_value()? {
                FlowModuleValue::AIAgent { omit_output_from_conversation, .. } => {
                    omit_output_from_conversation
                }
                _ => false,
            },
            _ => false,
        };

        let chat_ai_info = ChatAiInfo {
            chat_input_enabled: old_status.chat_input_enabled.unwrap_or(false),
            conversation_id: old_status.memory_id,
            is_ai_agent_step,
            omit_output_from_conversation,
        };
        (
            should_continue_flow,
            flow_job,
            flow_data,
            stop_early,
            skip_if_stop_early,
            nresult,
            is_failure_step,
            old_status.cleanup_module,
            chat_ai_info,
        )
    };

    let flow_job = Arc::new(flow_job);

    let done = if !should_continue_flow {
        {
            let logs = if flow_job.is_canceled() {
                "Flow job canceled\n".to_string()
            } else if stop_early {
                format!("Flow job stopped early because of a stop early predicate returning true\n")
            } else if is_failure_step {
                format!("Flow job completed with error, and error handler was triggered.\nIt completed with {}, and with recover: {}\n", if success { "success" } else { "error" }, result_has_recover_true(nresult.clone()))
            } else {
                format!(
                    "Flow job completed with {}\n",
                    if success { "success" } else { "error" }
                )
            };
            append_logs(&flow_job.id, w_id, logs, &db.into()).await;
        }
        #[cfg(feature = "enterprise")]
        if let Some(parent_job) = flow_job.parent_job {
            // if has parent job, append flow cleanup modules to parent job
            if !_cleanup_module.flow_jobs_to_clean.is_empty() {
                let uuids_json = serde_json::to_value(&_cleanup_module.flow_jobs_to_clean)
                    .map_err(|e| {
                        error::Error::internal_err(format!("Unable to serialize uuids: {e:#}"))
                    })?;
                sqlx::query!(
                    "UPDATE v2_job_status
                    SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)
                    WHERE id = $2",
                    uuids_json,
                    parent_job
                )
                .execute(db)
                .warn_after_seconds(3)
                .await?;
            }
            if !_cleanup_module.flow_jobs_to_schedule_clean.is_empty() {
                let entries_json = serde_json::to_value(
                    &_cleanup_module.flow_jobs_to_schedule_clean,
                )
                .map_err(|e| {
                    error::Error::internal_err(format!(
                        "Unable to serialize scheduled clean entries: {e:#}"
                    ))
                })?;
                sqlx::query!(
                    "UPDATE v2_job_status
                    SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)
                    WHERE id = $2",
                    entries_json,
                    parent_job
                )
                .execute(db)
                .warn_after_seconds(3)
                .await?;
            }
        } else {
            // run the cleanup step only when the root job is complete
            if !_cleanup_module.flow_jobs_to_clean.is_empty() {
                tracing::debug!(
                    "Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}",
                    _cleanup_module.flow_jobs_to_clean
                );
                sqlx::query!(
                    "UPDATE v2_job SET args = '{}'::jsonb WHERE id = ANY($1)",
                    &_cleanup_module.flow_jobs_to_clean,
                )
                .execute(db)
                .await
                .map_err(|e| {
                    Error::InternalErr(format!("error while cleaning up completed job: {e:#}"))
                })?;
                sqlx::query!(
                    "UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = ANY($1)",
                    &_cleanup_module.flow_jobs_to_clean,
                )
                .execute(db)
                .await
                .map_err(|e| {
                    Error::internal_err(format!("error while cleaning up completed job: {e:#}"))
                })?;
            }

            // Process scheduled deletions — insert into job_delete_schedule
            if !_cleanup_module.flow_jobs_to_schedule_clean.is_empty() {
                let w_id = &flow_job.workspace_id;
                for entry in &_cleanup_module.flow_jobs_to_schedule_clean {
                    windmill_common::jobs::schedule_job_deletion(
                        db,
                        entry.id,
                        w_id,
                        entry.delete_after_secs,
                    )
                    .await
                    .map_err(|e| {
                        Error::internal_err(format!(
                            "error scheduling job deletion for {}: {e:#}",
                            entry.id
                        ))
                    })?;
                }
            }

            // Flow-level delete_after_secs: apply to ALL child jobs + the parent flow job
            let flow_value = flow_data.value();
            let flow_delete_secs = windmill_common::jobs::resolve_delete_after_secs(
                flow_value.delete_after_use,
                flow_value.delete_after_secs,
            );
            if let Some(secs) = flow_delete_secs {
                let w_id = &flow_job.workspace_id;
                let mut all_ids: Vec<Uuid> =
                    sqlx::query_scalar!("SELECT id FROM v2_job WHERE root_job = $1", flow_job.id)
                        .fetch_all(db)
                        .await
                        .map_err(|e| {
                            Error::internal_err(format!(
                                "error fetching child jobs for flow-level deletion: {e:#}"
                            ))
                        })?;
                all_ids.push(flow_job.id);

                if secs == 0 {
                    sqlx::query!(
                        "UPDATE v2_job SET args = '{}'::jsonb WHERE id = ANY($1)",
                        &all_ids,
                    )
                    .execute(db)
                    .await
                    .map_err(|e| {
                        Error::internal_err(format!(
                            "error while cleaning up completed job args: {e:#}"
                        ))
                    })?;
                    sqlx::query!(
                        "UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = ANY($1)",
                        &all_ids,
                    )
                    .execute(db)
                    .await
                    .map_err(|e| {
                        Error::internal_err(format!(
                            "error while cleaning up completed job results: {e:#}"
                        ))
                    })?;
                    sqlx::query!(
                        "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = ANY($1)",
                        &all_ids,
                    )
                    .execute(db)
                    .await
                    .map_err(|e| {
                        Error::internal_err(format!(
                            "error while cleaning up completed job logs: {e:#}"
                        ))
                    })?;
                } else {
                    for id in &all_ids {
                        windmill_common::jobs::schedule_job_deletion(db, *id, w_id, secs)
                            .await
                            .map_err(|e| {
                                Error::internal_err(format!(
                                    "error scheduling job deletion for {id}: {e:#}"
                                ))
                            })?;
                    }
                }
            }
        }

        if flow_job.is_canceled() {
            let canceled_by = CanceledBy {
                username: flow_job.canceled_by.clone(),
                reason: flow_job.canceled_reason.clone(),
            };
            let error = canceled_job_to_result(&flow_job);
            add_completed_job_error(
                db,
                &MiniCompletedJob::from(flow_job.clone()),
                0,
                Some(canceled_by),
                error,
                worker_name,
                true,
                None,
            )
            .await?;
        } else {
            let cflow_job: MiniCompletedJob = MiniCompletedJob::from(flow_job.clone());

            if flow_job.cache_ttl.is_some() && success {
                let flow = RawData::Flow(flow_data.clone());
                let cached_res_path = cached_result_path(db, client, &flow_job, Some(&flow)).await;

                save_in_cache(
                    db,
                    client,
                    &MiniCompletedJob::from(cflow_job.clone()),
                    cached_res_path,
                    nresult.clone(),
                )
                .await;
            }

            let success = success && (!is_failure_step || result_has_recover_true(nresult.clone()));

            add_time!(bench, "flow status update 1");

            let skipped = stop_early && skip_if_stop_early;
            add_tool_message_to_conversation(
                db,
                &job_id_for_status,
                success,
                skipped,
                chat_ai_info.is_ai_agent_step,
                chat_ai_info.omit_output_from_conversation,
                &nresult,
                chat_ai_info.chat_input_enabled,
                chat_ai_info.conversation_id,
            )
            .await?;
            let (duration, wac_job_ids) = if success {
                let (_, duration, wac_job_ids) = add_completed_job(
                    db,
                    &cflow_job,
                    true,
                    skipped,
                    Json(&nresult),
                    None,
                    0,
                    None,
                    true,
                    None,
                    false,
                )
                .await?;
                (duration, wac_job_ids)
            } else {
                let (_, duration, wac_job_ids) = add_completed_job(
                    db,
                    &cflow_job,
                    false,
                    skipped,
                    Json(
                        &serde_json::from_str::<Value>(nresult.get()).unwrap_or_else(
                            |e| json!({"error": format!("Impossible to serialize error: {e:#}")}),
                        ),
                    ),
                    None,
                    0,
                    canceled_by.clone(),
                    true,
                    None,
                    false,
                )
                .await?;
                (duration, wac_job_ids)
            };
            flow_job_duration = flow_job
                .started_at
                .map(|x| FlowJobDuration { started_at: x, duration_ms: duration });

            // If this flow is a WAC child (not a flow step, has parent),
            // notify the WAC parent of completion.
            if !flow_job.is_flow_step() {
                if let Some(parent_job) = flow_job.parent_job {
                    if let Some(job_ids) = wac_job_ids {
                        let _ = crate::result_processor::handle_wac_child_completion(
                            db,
                            &flow_job.id,
                            parent_job,
                            &flow_job.workspace_id,
                            nresult.clone(),
                            success,
                            job_ids,
                        )
                        .await;
                    }
                }
            }
        }
        true
    } else {
        tracing::debug!(id = %flow_job.id,  "start handle flow");
        match Box::pin(handle_flow(
            flow_job.clone(),
            &flow_data,
            db,
            client,
            Some(nresult.clone()),
            same_worker_tx,
            worker_dir,
            job_completed_tx,
            worker_name,
            flow_runners,
            &killpill_rx,
        ))
        .warn_after_seconds(10)
        .await
        {
            Err(err) => {
                let e = json!({"message": err.to_string(), "name": "InternalError"});
                append_logs(
                    &flow_job.id,
                    w_id,
                    format!("Unexpected error during flow chaining:\n{:#?}", e),
                    &db.into(),
                )
                .await;
                let _ = add_completed_job_error(
                    db,
                    &MiniCompletedJob::from(flow_job.clone()),
                    0,
                    None,
                    e,
                    worker_name,
                    true,
                    None,
                )
                .await;
                // override prior child's success so the parent learns this flow failed
                success = false;
                true
            }
            Ok(_) => false,
        }
    };

    if done {
        if flow_job.same_worker && !KEEP_JOB_DIR.load(Ordering::Relaxed) {
            let _ = tokio::fs::remove_dir_all(format!("{worker_dir}/{}", flow_job.id)).await;
        }

        if flow_job.is_flow_step() {
            if let Some(parent_job) = flow_job.parent_job {
                tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status");

                return Ok(UpdateFlowStatusAfterJobCompletion::Rec(
                    RecUpdateFlowStatusAfterJobCompletion {
                        flow: parent_job,
                        job_id_for_status: flow,
                        success: success && !is_failure_step,
                        canceled_by: if !success { canceled_by.clone() } else { None },
                        flow_job_duration: flow_job_duration.clone(),
                        result: nresult.clone(),
                        stop_early_override: if stop_early {
                            Some(skip_if_stop_early)
                        } else {
                            None
                        },
                        has_triggered_error_handler: has_triggered_error_handler || is_failure_step,
                    },
                ));
            }
        }
        Ok(UpdateFlowStatusAfterJobCompletion::Done(flow_job))
    } else {
        Ok(UpdateFlowStatusAfterJobCompletion::NotDone)
    }
}

fn find_flow_job_index(flow_jobs: &Vec<Uuid>, job_id_for_status: &Uuid) -> Option<usize> {
    flow_jobs.iter().position(|x| x == job_id_for_status)
}

fn format_chat_message_value(value: &Value) -> String {
    match value {
        Value::Null => "null".to_string(),
        Value::Bool(_) | Value::Number(_) => value.to_string(),
        Value::String(text) => text.clone(),
        Value::Array(_) | Value::Object(_) => serde_json::to_string_pretty(value)
            .unwrap_or_else(|e| format!("Failed to serialize result: {e}")),
    }
}

/// Selects the assistant message persisted for the final non-AI step in a chat-enabled flow.
/// `windmill_chat_answer` is the explicit override contract:
/// - `null` suppresses the assistant message
/// - any other JSON value is rendered as the chat message
fn extract_chat_message_from_flow_result(result: &RawValue) -> error::Result<Option<String>> {
    let value: Value = serde_json::from_str(result.get())
        .map_err(|e| Error::internal_err(format!("Failed to parse flow result: {e}")))?;

    match value {
        Value::Object(map) => {
            match map.get("windmill_chat_answer") {
                Some(Value::Null) => return Ok(None),
                Some(answer) => return Ok(Some(format_chat_message_value(answer))),
                _ => {}
            }

            Ok(Some(
                serde_json::to_string_pretty(&Value::Object(map))
                    .unwrap_or_else(|e| format!("Failed to serialize result: {e}")),
            ))
        }
        Value::String(content) => Ok(Some(content)),
        value => Ok(Some(
            serde_json::to_string_pretty(&value)
                .unwrap_or_else(|e| format!("Failed to serialize result: {e}")),
        )),
    }
}

async fn add_tool_message_to_conversation(
    db: &DB,
    job_id: &Uuid,
    success: bool,
    skipped: bool,
    is_ai_agent_step: bool,
    omit_output_from_conversation: bool,
    result: &Box<RawValue>,
    chat_input_enabled: bool,
    conversation_id: Option<Uuid>,
) -> error::Result<()> {
    if is_ai_agent_step && omit_output_from_conversation {
        return Ok(());
    }

    // Create assistant message if it's a flow and it's done, but only if last module is not an AI agent
    if !skipped && chat_input_enabled {
        // Get conversation_id from flow_status.memory_id

        if let Some(conversation_id) = conversation_id {
            // Only create assistant message if last module is NOT an AI agent, or there was an error
            if !is_ai_agent_step || success == false {
                let Some(content) = extract_chat_message_from_flow_result(result.as_ref())? else {
                    return Ok(());
                };

                // Insert new assistant message
                let mut tx = db.begin().await?;
                add_message_to_conversation_tx(
                    &mut tx,
                    conversation_id,
                    Some(job_id.clone()),
                    &content,
                    MessageType::Assistant,
                    None,
                    success,
                )
                .await?;
                tx.commit().await?;
            }
        }
    }

    Ok(())
}

async fn set_success_and_duration_in_flow_job_success<'c>(
    flow_jobs_success: &Option<Vec<Option<bool>>>,
    flow_jobs: &Vec<Uuid>,
    job_id_for_status: &Uuid,
    old_status_step: i32,
    flow: Uuid,
    success: bool,
    flow_job_duration: Option<FlowJobDuration>,
    tx: &mut Transaction<'c, Postgres>,
) -> error::Result<()> {
    if flow_jobs_success.is_some() {
        let position = find_flow_job_index(flow_jobs, job_id_for_status);
        if let Some(position) = position {
            sqlx::query!(
                "UPDATE v2_job_status SET flow_status = 
            CASE WHEN flow_status->'modules'->$1::int->'flow_jobs_duration' IS NOT NULL THEN
                     JSONB_SET(JSONB_SET(JSONB_SET(
                         flow_status,
                         ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],
                         $4
                     ),
                     ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5),
                     ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6)
                 ELSE
                    JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4)
            END
            WHERE id = $2",
                old_status_step as i32,
                flow,
                position as i32,
                json!(success),
                flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(),
                flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default()
            )
            .execute(&mut **tx)
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "error while setting flow_jobs_success/timeline: {e:#}"
                ))
            })?;
        }
    }
    Ok(())
}

async fn retrieve_flow_jobs_results(
    db: &DB,
    w_id: &str,
    job_uuids: &Vec<Uuid>,
) -> error::Result<Box<RawValue>> {
    let results = sqlx::query!(
        "SELECT result, id
         FROM v2_job_completed
         WHERE id = ANY($1) AND workspace_id = $2",
        job_uuids.as_slice(),
        w_id
    )
    .fetch_all(db)
    .await?
    .into_iter()
    .map(|br| (br.id, br.result))
    .collect::<HashMap<_, _>>();

    let results = job_uuids
        .iter()
        .map(|j| {
            results
                .get(j)
                .ok_or_else(|| Error::internal_err(format!("missing job result for {}", j)))
        })
        .collect::<Result<Vec<_>, _>>()?;
    tracing::debug!("Retrieved results for flow jobs {:?}", results);
    Ok(to_raw_value(&results))
}

async fn compute_skip_branchall_failure<'c>(
    branch: usize,
    parallel: bool,
    flow_module: Option<&FlowModule>,
    successes: &Option<Vec<Option<bool>>>,
) -> windmill_common::error::Result<bool> {
    let branches = flow_module
        .and_then(|x| x.get_branches_skip_failures().ok())
        .map(|x| {
            x.branches
                .iter()
                .map(|b| b.skip_failure.unwrap_or(false))
                .collect::<Vec<_>>()
        });
    if parallel {
        if let Some(successes) = successes {
            for (i, success) in successes.iter().enumerate() {
                if branches
                    .as_ref()
                    .and_then(|x| x.get(i))
                    .unwrap_or(&false)
                    .to_owned()
                {
                    continue;
                }
                if !(success.unwrap_or(false)) {
                    return Ok(false);
                }
            }
            Ok(true)
        } else {
            Ok(false)
        }
    } else {
        Ok(branches
            .and_then(|x| x.get(branch as usize).map(|b| b.to_owned()))
            .unwrap_or(false))
    }
}

// async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result<FlowCleanupModule, Error> {
//     tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid);
//     let raw_value = sqlx::query_scalar!(
//         "SELECT flow_status->'cleanup_module' as cleanup_module
//         FROM queue
//         WHERE id = $1",
//         flow_uuid,
//     )
//     .fetch_one(db)
//     .await
//     .map_err(|e| Error::internal_err(format!("error during retrieval of cleanup module: {e:#}")))?;

//     raw_value
//         .clone()
//         .and_then(|rv| serde_json::from_value::<FlowCleanupModule>(rv).ok())
//         .ok_or(Error::internal_err(format!(
//             "Unable to parse flow cleanup module {:?}",
//             raw_value
//         )))
// }

async fn evaluate_retry(
    retry: &Retry,
    status: &RetryStatus,
    result: Arc<Box<RawValue>>,
    flow_args: Marc<HashMap<String, Box<RawValue>>>,
    flow_env: Option<&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,
            flow_env,
            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(
    expr: &str,
    flow_args: Marc<HashMap<String, Box<RawValue>>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    result: Arc<Box<RawValue>>,
    all_iters: Option<Arc<Box<RawValue>>>,
    by_id: Option<&IdContext>,
    client: Option<&AuthedClient>,
    resumes: Option<(Arc<Box<RawValue>>, Arc<Box<RawValue>>, Arc<Box<RawValue>>)>,
    ctx: Option<Vec<(String, String)>>,
) -> error::Result<bool> {
    let mut context = HashMap::with_capacity(if resumes.is_some() { 7 } else { 3 });
    context.insert("result".to_string(), result.clone());
    if let Some(all_iters) = all_iters {
        context.insert("all_iters".to_string(), all_iters);
    }
    context.insert("previous_result".to_string(), result.clone());

    if let Some(resumes) = resumes {
        context.insert("resume".to_string(), resumes.1);
        context.insert("resumes".to_string(), resumes.0);
        context.insert("approvers".to_string(), resumes.2);
    }

    match eval_timeout(
        format!("Boolean({expr})"),
        context,
        Some(flow_args),
        flow_env,
        client,
        by_id,
        ctx,
    )
    .await?
    .get()
    {
        "true" => Ok(true),
        "false" => Ok(false),
        a @ _ => Err(Error::ExecutionErr(format!(
            "Expected a boolean value, found: {a:?}"
        ))),
    }
}

// Appends a predicate-evaluation error to the root flow job's logs so users can
// see which boolean expression failed even when the failure is masked further up
// the flow tree (e.g. by `skip_failures: true` on a surrounding forloop).
async fn append_predicate_error_to_root_logs(
    db: &DB,
    root_flow_id: Uuid,
    workspace_id: &str,
    module_id: Option<&str>,
    predicate: &str,
    expr: &str,
    err: &Error,
) {
    let module_part = module_id
        .map(|id| format!(" of module '{id}'"))
        .unwrap_or_default();
    let log = format!("Error evaluating {predicate}{module_part} expression `{expr}`: {err:#}\n");
    append_logs(&root_flow_id, workspace_id, log, &db.into()).await;
}

fn root_flow_id_for(flow_job: &MiniPulledJob) -> Uuid {
    flow_job.flow_innermost_root_job.unwrap_or(flow_job.id)
}

async fn fetch_root_flow_id(db: &DB, flow_id: Uuid) -> Uuid {
    sqlx::query_scalar!(
        "SELECT flow_innermost_root_job FROM v2_job WHERE id = $1",
        flow_id
    )
    .fetch_optional(db)
    .await
    .ok()
    .flatten()
    .flatten()
    .unwrap_or(flow_id)
}

// Resolve the flow_env to use for predicate evaluations in
// `update_flow_status_after_job_completion_internal`: take the current flow's
// `flow_env` if present, otherwise inherit from the root flow, then interpolate
// any `$var:`/`$res:` references via `transform_json`.
//
// Returns `Some((env, is_cacheable))`. `is_cacheable` is `false` only when the
// returned env is a partially-resolved fallback after a transient error (e.g.
// `transform_json` failed mid-resolve, or the mini job fetch failed). Callers
// must not cache `is_cacheable == false` results — doing so would freeze the
// transient failure for the rest of the flow run.
async fn resolve_flow_env_for_status_update(
    db: &DB,
    client: &AuthedClient,
    flow_job_id: Uuid,
    workspace_id: &str,
    flow_value: &FlowValue,
) -> Option<(HashMap<String, Box<RawValue>>, bool)> {
    // Fetch the env source. For the inherited path, we first need to know whether the
    // flow even has a parent — `fetch_root_flow_env` runs a recursive CTE on `v2_job`
    // and is wasted work for top-level flows with no own `flow_env`. The mini job we
    // pull here can be reused below if `transform_json` ends up needing it.
    let (env, mini): (HashMap<String, Box<RawValue>>, Option<MiniPulledJob>) =
        if let Some(ref e) = flow_value.flow_env {
            (e.clone(), None)
        } else {
            let mini = match get_mini_pulled_job(db, &flow_job_id).await {
                Ok(Some(j)) => j,
                Ok(None) => return None,
                Err(e) => {
                    tracing::warn!("Failed to fetch flow job to resolve flow_env: {e:#}");
                    return None;
                }
            };
            if mini.parent_job.is_none() {
                // No own flow_env and no parent to inherit from — nothing to resolve.
                return None;
            }
            let env = fetch_root_flow_env(db, flow_job_id, workspace_id).await?;
            (env, Some(mini))
        };
    if env.is_empty() {
        return Some((env, true));
    }
    // Skip the DB roundtrip + `transform_json` when nothing in env needs interpolation.
    // This is the common case: a flow_env containing only literal values.
    if !crate::common::map_needs_resolution(&env) {
        return Some((env, true));
    }
    let mini = match mini {
        Some(m) => m,
        None => match get_mini_pulled_job(db, &flow_job_id).await {
            Ok(Some(j)) => j,
            // Don't cache: we couldn't fetch the job, so the env we'd return is the
            // pre-interpolation literal — caching that would freeze the broken state.
            Ok(None) => return Some((env, false)),
            Err(e) => {
                tracing::warn!("Failed to fetch flow job to resolve flow_env: {e:#}");
                return Some((env, false));
            }
        },
    };
    match transform_json(
        client,
        workspace_id,
        &env,
        &mini,
        &Connection::Sql(db.clone()),
    )
    .await
    {
        Ok(Some(resolved)) => Some((resolved, true)),
        Ok(None) => Some((env, true)),
        // Don't cache transient resolution failures: one variable/resource API blip
        // would otherwise poison the literal-only env for the rest of the flow run.
        Err(e) => {
            tracing::warn!("Failed to resolve flow_env references in status update: {e:#}");
            Some((env, false))
        }
    }
}

// Maximum number of ancestor jobs the flow_env walk will follow before bailing.
// Bounds runtime cost and protects against pathological data.
const MAX_FLOW_ENV_LOOKUP_DEPTH: i32 = 50;

// Look up the nearest ancestor flow's `flow_env` for the given flow job.
// Sub-flows spawned by branches/loops via `payload_from_modules` don't carry
// `flow_env` in their own `FlowValue`, so any predicate evaluated against the
// local flow_env would see `None`. We walk the ancestor chain via
// `flow_innermost_root_job → parent_job` (one step at a time) and return the
// CLOSEST ancestor whose persisted flow definition (`flow_version.value` or
// `raw_flow`) carries `flow_env`.
//
// The walk deliberately does NOT use `root_job` (which would jump straight to
// the topmost parent) because imported flows define their own `flow_env`
// scope: a branch inside an imported flow must see the imported flow's env,
// not the top parent's. `flow_innermost_root_job` walks one scope at a time
// and resets to NULL on parallel-loop iterations and imported sub-flows, so
// recursion is required to walk past those resets to the closest scope.
async fn fetch_root_flow_env(
    db: &DB,
    flow_job_id: Uuid,
    workspace_id: &str,
) -> Option<HashMap<String, Box<RawValue>>> {
    sqlx::query_scalar!(
        r#"WITH RECURSIVE chain(id, parent_job, flow_innermost_root_job, runnable_id, runnable_path, raw_flow, depth) AS (
            SELECT id, parent_job, flow_innermost_root_job, runnable_id, runnable_path, raw_flow, 0
            FROM v2_job
            WHERE id = $1 AND workspace_id = $2
            UNION ALL
            SELECT j.id, j.parent_job, j.flow_innermost_root_job, j.runnable_id, j.runnable_path, j.raw_flow, c.depth + 1
            FROM v2_job j
            JOIN chain c
                ON j.id = COALESCE(c.flow_innermost_root_job, c.parent_job)
            WHERE j.workspace_id = $2 AND c.depth < $3
        )
        SELECT
            CASE
                WHEN flow_version.id IS NOT NULL THEN
                    flow_version.value -> 'flow_env'
                ELSE
                    chain.raw_flow -> 'flow_env'
            END AS "flow_env: Json<HashMap<String, Box<RawValue>>>"
         FROM chain
         LEFT JOIN flow_version
             ON flow_version.id = chain.runnable_id
            AND flow_version.path = chain.runnable_path
            AND flow_version.workspace_id = $2
         WHERE (CASE
                WHEN flow_version.id IS NOT NULL THEN flow_version.value -> 'flow_env'
                ELSE chain.raw_flow -> 'flow_env'
            END) IS NOT NULL
         ORDER BY chain.depth ASC
         LIMIT 1"#,
        flow_job_id,
        workspace_id,
        MAX_FLOW_ENV_LOOKUP_DEPTH,
    )
    .fetch_optional(db)
    .await
    .ok()
    .flatten()
    .flatten()
    .map(|json| json.0)
}

struct FailureContext {
    started_at: Arc<Box<RawValue>>,
    flow_job_id: Uuid,
}

#[instrument(level = "trace", skip_all)]
pub async fn evaluate_input_transform<T>(
    transform: &InputTransform,
    last_result: Arc<Box<RawValue>>,
    flow_args: Option<Marc<HashMap<String, Box<RawValue>>>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    authed_client: Option<&AuthedClient>,
    by_id: Option<&IdContext>,
) -> error::Result<T>
where
    T: for<'de> serde::Deserialize<'de> + Send + Default,
{
    match transform {
        InputTransform::Static { value } => serde_json::from_str(value.get()).map_err(|e| {
            Error::ExecutionErr(format!(
                "Error parsing static value as {}: {e:#}",
                std::any::type_name::<T>()
            ))
        }),
        InputTransform::Javascript { expr } => {
            let mut context = HashMap::with_capacity(2);
            context.insert("result".to_string(), last_result.clone());
            context.insert("previous_result".to_string(), last_result.clone());
            let result = eval_timeout(
                expr.to_string(),
                context,
                flow_args,
                flow_env,
                authed_client,
                by_id,
                None,
            )
            .warn_after_seconds(3)
            .await
            .map_err(|e| {
                Error::ExecutionErr(format!(
                    "Error during evaluation of expression `{expr}`:\n{e:#}"
                ))
            })?;

            serde_json::from_str(result.get()).map_err(|e| {
                Error::ExecutionErr(format!(
                    "Error parsing result as {}: {e:#}. Value was: {}",
                    std::any::type_name::<T>(),
                    result.get()
                ))
            })
        }
        InputTransform::Ai => Ok(T::default()),
    }
}

/// resumes should be in order of timestamp ascending, so that more recent are at the end
#[instrument(level = "trace", skip_all)]
async fn transform_input(
    flow_args: Marc<HashMap<String, Box<RawValue>>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    last_result: Arc<Box<RawValue>>,
    input_transforms: &HashMap<String, InputTransform>,
    resumes: Arc<Box<RawValue>>,
    resume: Arc<Box<RawValue>>,
    approvers: Arc<Box<RawValue>>,
    failure_context: Option<FailureContext>,
    by_id: &IdContext,
    client: &AuthedClient,
) -> windmill_common::error::Result<HashMap<String, Box<RawValue>>> {
    let mut mapped = HashMap::new();

    let mut env = HashMap::new();

    if input_transforms
        .iter()
        .any(|x| matches!(x.1, InputTransform::Javascript { .. }))
    {
        env.insert("params".to_string(), Arc::new(to_raw_value(&mapped)));
        env.insert("previous_result".to_string(), last_result);
        env.insert("resume".to_string(), resume);
        env.insert("resumes".to_string(), resumes);
        env.insert("approvers".to_string(), approvers);
        if let Some(FailureContext { started_at, flow_job_id }) = failure_context {
            env.insert("started_at".to_string(), started_at);
            env.insert(
                "flow_job_id".to_string(),
                Arc::new(to_raw_value(&flow_job_id)),
            );
        }
    }

    for (key, val) in input_transforms.into_iter() {
        match val {
            InputTransform::Static { value } => {
                mapped.insert(key.to_string(), to_raw_value(&value));
            }
            InputTransform::Javascript { expr } => {
                let v = eval_timeout(
                    expr.to_string(),
                    env.clone(),
                    Some(flow_args.clone()),
                    flow_env,
                    Some(client),
                    Some(by_id),
                    None,
                )
                .await
                .map_err(|e| {
                    Error::ExecutionErr(format!(
                        "Error during isolated evaluation of expression `{expr}`:\n{e:#}"
                    ))
                })?;
                mapped.insert(key.to_string(), v);
            }
            InputTransform::Ai => (),
        }
    }

    Ok(mapped)
}

#[instrument(level = "trace", skip_all)]
pub async fn handle_flow(
    flow_job: Arc<MiniPulledJob>,
    flow_data: &cache::FlowData,
    db: &sqlx::Pool<sqlx::Postgres>,
    client: &AuthedClient,
    last_result: Option<Arc<Box<RawValue>>>,
    same_worker_tx: &SameWorkerSender,
    worker_dir: &str,
    job_completed_tx: JobCompletedSender,
    worker_name: &str,
    flow_runners: Option<Arc<FlowRunners>>,
    killpill_rx: &tokio::sync::broadcast::Receiver<()>,
) -> anyhow::Result<()> {
    let flow = flow_data.value();

    // Sub-flows spawned by `payload_from_modules` for branches/loops don't
    // carry the parent's `flow_env` in their own FlowValue. Fall back to the
    // nearest enclosing scope's `flow_env` so predicates like `skip_if`,
    // `stop_after_if`, and branch conditions see the same env as input
    // transforms.
    let inherited_env: Option<HashMap<String, Box<RawValue>>> =
        if flow.flow_env.is_none() && flow_job.parent_job.is_some() {
            fetch_root_flow_env(db, flow_job.id, &flow_job.workspace_id).await
        } else {
            None
        };
    let env_source = flow.flow_env.as_ref().or(inherited_env.as_ref());

    // Resolve $var: and $res: references in flow_env.
    // We resolve into a separate variable to avoid cloning the entire FlowValue
    // (which includes modules, failure_module, etc.) just to replace flow_env.
    // `is_cacheable` is `false` only on a transient `transform_json` failure,
    // matching `resolve_flow_env_for_status_update`'s contract — we must not
    // freeze a partial resolution into the cache.
    let resolved_env;
    let (flow_env, is_cacheable) = if let Some(env) = env_source {
        match transform_json(
            client,
            &flow_job.workspace_id,
            env,
            &flow_job,
            &Connection::Sql(db.clone()),
        )
        .await
        {
            Ok(Some(resolved)) => {
                resolved_env = resolved;
                (Some(&resolved_env), true)
            }
            Ok(None) => (Some(env), true),
            Err(e) => {
                tracing::warn!("Failed to resolve flow_env references: {e}");
                (Some(env), false)
            }
        }
    } else {
        (None, true)
    };

    // Populate the per-flow resolved-env cache so subsequent predicate
    // evaluations in `update_flow_status_after_job_completion_internal` skip
    // the recursive CTE + transform_json. Costs one HashMap clone per
    // sub-flow handle_flow entry; saves up to one CTE + one transform_json
    // per flow execution that has predicates referencing flow_env.
    if is_cacheable {
        if let Some(env) = flow_env {
            if !env.is_empty() {
                RESOLVED_FLOW_ENV_CACHE.insert(flow_job.id, Arc::new(env.clone()));
            }
        }
    }

    let status = flow_job
        .parse_flow_status()
        .with_context(|| "Unable to parse flow status")?;

    let schedule_path = flow_job.schedule_path();
    if !flow_job.is_flow_step()
        && status.retry.fail_count == 0
        && schedule_path.is_some()
        && flow_job.runnable_path.is_some()
        && status.step == 0
    {
        let schedule_path = schedule_path.as_ref().unwrap();

        let schedule = get_schedule_opt(db, &flow_job.workspace_id, schedule_path)
            .warn_after_seconds(5)
            .await?;

        if let Some(schedule) = schedule {
            let mini_job = MiniCompletedJob::from(flow_job.clone());
            let runnable_path = flow_job.runnable_path.as_ref().unwrap().clone();
            let schedule_push_result = (|| async {
                let tx = db.begin().warn_after_seconds(5).await
                    .map_err(|e| Error::internal_err(format!("begin tx for schedule push: {e:#}")))?;
                let (tx, schedule_push_err) = try_schedule_next_job(
                    db,
                    tx,
                    &mini_job,
                    &schedule,
                    &runnable_path,
                )
                .await;
                if let Some(err) = schedule_push_err {
                    return Err(err);
                }
                tx.commit().warn_after_seconds(5).await
                    .map_err(|e| Error::internal_err(format!("commit schedule push: {e:#}")))?;
                Ok::<(), Error>(())
            })
            .retry(
                ConstantBuilder::default()
                    .with_delay(std::time::Duration::from_secs(3))
                    .with_max_times(10)
                    .build(),
            )
            .when(|err: &Error| !matches!(err, Error::QuotaExceeded(_) | Error::NotFound(_)))
            .notify(|err: &Error, dur: std::time::Duration| {
                tracing::error!(
                    "Could not push next scheduled job for flow schedule {}, retrying in {dur:#?}: {err:#?}",
                    schedule.path
                );
            })
            .sleep(tokio::time::sleep)
            .await;

            // Non-retryable errors (QuotaExceeded, NotFound) are handled inside
            // try_schedule_next_job (schedule disabled, returns None), so they never
            // reach here. This handles only transient errors after retry exhaustion.
            if let Err(err) = schedule_push_result {
                tracing::error!(
                    "Could not push next scheduled job for {} after retries: {err}. Disabling schedule.",
                    schedule.path
                );
                if let Err(disable_err) = sqlx::query!(
                    "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
                    err.to_string(),
                    &flow_job.workspace_id,
                    &schedule.path
                )
                .execute(db)
                .await
                {
                    report_error_to_workspace_handler_or_critical_side_channel(
                        &mini_job,
                        db,
                        format!(
                            "Could not push next scheduled job for {} and could not disable schedule: {disable_err}",
                            schedule.path,
                        ),
                    )
                    .await;
                    return Err(SchedulePushZombieError(
                        format!(
                            "Could not push or disable schedule {} after retries",
                            schedule.path
                        ),
                    ).into());
                }
            }
        } else {
            tracing::error!(
                "Schedule {schedule_path} in {} not found. Impossible to schedule again",
                &flow_job.workspace_id
            );
        }
    }

    let mut rec = PushNextFlowJobRec { flow_job: flow_job, status: status };
    loop {
        let PushNextFlowJobRec { flow_job, status } = rec;
        let next = Box::pin(push_next_flow_job(
            flow_job,
            status,
            flow,
            flow_env,
            db,
            client,
            last_result.clone(),
            same_worker_tx,
            worker_dir,
            worker_name,
            flow_runners.clone(),
            job_completed_tx.clone(),
            &killpill_rx,
        ))
        .warn_after_seconds(10)
        .await?;
        match next {
            PushNextFlowJob::Rec(nrec) => {
                tracing::info!("recursively pushing next flow job {}", nrec.flow_job.id);
                rec = nrec;
            }
            PushNextFlowJob::Done(update_flow) => {
                if let Some(update_flow) = update_flow {
                    tracing::info!(
                        "sending flow status update {} with success {} to job completed channel",
                        update_flow.flow,
                        update_flow.success
                    );
                    job_completed_tx
                        .send(SendResultPayload::UpdateFlow(update_flow), false)
                        .warn_after_seconds(3)
                        .await
                        .map_err(|e| {
                            Error::internal_err(format!(
                                "error sending update flow message to job completed channel: {e:#}"
                            ))
                        })?;
                }
                break;
            }
        }
    }

    Ok(())
}

#[derive(Serialize, Debug)]
pub struct Iter {
    index: i32,
    value: Box<RawValue>,
}

#[derive(FromRow)]
pub struct ResumeRow {
    pub value: Json<Box<RawValue>>,
    pub approver: Option<String>,
    pub approved: bool,
    pub resume_id: i32,
}

lazy_static::lazy_static! {
    static ref CRASH_FORCEFULLY_AT_STEP: Option<usize> = std::env::var("CRASH_FORCEFULLY_AT_STEP")
        .ok()
        .and_then(|x| x.parse::<usize>().ok());

    static ref CRASH_STEP_COUNTER: AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

    static ref BRANCHALL_INDEX_RE: regex::Regex = regex::Regex::new(r"/branchall-(\d+)$").unwrap();
}

#[inline(always)]
fn potentially_crash_for_testing() {
    #[cfg(feature = "flow_testing")]
    if let Some(crash_at) = CRASH_FORCEFULLY_AT_STEP.as_ref() {
        let counter = CRASH_STEP_COUNTER.fetch_add(1, Ordering::SeqCst);
        if &counter == crash_at {
            panic!("CRASH#1 - expected crash for testing at step {}", crash_at);
        }
    }
}

// static
lazy_static::lazy_static! {
    pub static ref EHM: HashMap<String, Box<RawValue>> = HashMap::new();
}

#[derive(Debug)]
enum PushNextFlowJob {
    Rec(PushNextFlowJobRec),
    Done(Option<UpdateFlow>),
}

#[derive(Debug)]
struct PushNextFlowJobRec {
    flow_job: Arc<MiniPulledJob>,
    status: FlowStatus,
}

// #[async_recursion]
// #[instrument(level = "trace", skip_all)]
/// Resolve the worker tag for a flow's child job (step, nested sub-flow, preprocessor).
///
/// A child normally inherits the parent flow job's tag so the whole flow runs on one worker
/// group. The exceptions, in order:
/// - the preprocessor step, or a flow running on the generic `flow` / `flow-{workspace}` tag,
///   always uses the child's own tag (`step_tag`);
/// - when the flow opts into `preserve_step_tags` and the child declares its own non-empty tag,
///   that tag is honored instead of being overridden by the flow tag;
/// - otherwise the child inherits the parent flow job's tag.
fn resolve_flow_step_tag(
    is_preprocessor_step: bool,
    flow_tag: &str,
    workspace_id: &str,
    preserve_step_tags: bool,
    step_tag: Option<&str>,
) -> Option<String> {
    if is_preprocessor_step || flow_tag == "flow" || flow_tag == format!("flow-{}", workspace_id) {
        step_tag.map(str::to_string)
    } else if preserve_step_tags && step_tag.is_some_and(|t| !t.is_empty()) {
        step_tag.map(str::to_string)
    } else {
        Some(flow_tag.to_string())
    }
}

#[cfg(test)]
mod tag_resolution_tests {
    use super::resolve_flow_step_tag;

    #[test]
    fn step_inherits_custom_flow_tag_by_default() {
        // Parent flow on a custom tag, step declares its own tag, preserve disabled:
        // the step inherits the flow tag (historical behavior).
        assert_eq!(
            resolve_flow_step_tag(false, "worker-group-A", "w1", false, Some("worker-group-B")),
            Some("worker-group-A".to_string())
        );
    }

    #[test]
    fn step_keeps_own_tag_when_preserve_enabled() {
        // The exact customer scenario: a sub-flow tagged worker-group-B run as a step of a
        // flow tagged worker-group-A now runs on worker-group-B when preserve_step_tags is on.
        assert_eq!(
            resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("worker-group-B")),
            Some("worker-group-B".to_string())
        );
    }

    #[test]
    fn untagged_step_inherits_flow_tag_even_when_preserve_enabled() {
        assert_eq!(
            resolve_flow_step_tag(false, "worker-group-A", "w1", true, None),
            Some("worker-group-A".to_string())
        );
        // An empty tag counts as "no tag" and still inherits.
        assert_eq!(
            resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("")),
            Some("worker-group-A".to_string())
        );
    }

    #[test]
    fn generic_flow_tag_always_uses_step_tag() {
        for flow_tag in ["flow", "flow-w1"] {
            assert_eq!(
                resolve_flow_step_tag(false, flow_tag, "w1", false, Some("worker-group-B")),
                Some("worker-group-B".to_string())
            );
        }
    }

    #[test]
    fn preprocessor_step_uses_step_tag() {
        assert_eq!(
            resolve_flow_step_tag(true, "worker-group-A", "w1", false, Some("worker-group-B")),
            Some("worker-group-B".to_string())
        );
    }
}

async fn push_next_flow_job(
    flow_job: Arc<MiniPulledJob>,
    mut status: FlowStatus,
    flow: &FlowValue,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    db: &sqlx::Pool<sqlx::Postgres>,
    client: &AuthedClient,
    last_job_result: Option<Arc<Box<RawValue>>>,
    same_worker_tx: &SameWorkerSender,
    worker_dir: &str,
    worker_name: &str,
    flow_runners: Option<Arc<FlowRunners>>,
    job_completed_tx: JobCompletedSender,
    killpill_rx: &tokio::sync::broadcast::Receiver<()>,
) -> error::Result<PushNextFlowJob> {
    let job_root = flow_job
        .flow_innermost_root_job
        .map(|x| x.to_string())
        .unwrap_or_else(|| "none".to_string());

    let mut step = Step::from_i32_and_len(status.step, flow.modules.len());

    tracing::info!(id = %flow_job.id, root_id = %job_root, step = ?step, "pushing next flow job");

    let mut status_module = match step {
        Step::Step { idx: i, .. } => status
            .modules
            .get(i)
            .cloned()
            .unwrap_or_else(|| status.failure_module.module_status.clone()),
        Step::PreprocessorStep => status
            .preprocessor_module
            .clone()
            .unwrap_or_else(|| status.failure_module.module_status.clone()),
        Step::FailureStep => status.failure_module.module_status.clone(),
    };

    // tracing::error!("status_module: {status_module:#?}");

    let fj: mappable_rc::Marc<MiniPulledJob> = flow_job.clone().into();
    let arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>> =
        Marc::map(fj, |x: &MiniPulledJob| {
            if let Some(args) = &x.args {
                &args.0
            } else {
                &EHM
            }
        });

    // if this is an empty module without preprocessor or if the module has already been completed, successfully, update the parent flow
    if (flow.modules.is_empty() && !step.is_preprocessor_step())
        || matches!(status_module, FlowStatusModule::Success { .. })
    {
        return Ok(PushNextFlowJob::Done(Some(UpdateFlow {
            flow: flow_job.id,
            success: true,
            result: if flow.modules.is_empty() {
                to_raw_value(arc_flow_job_args.as_ref())
            } else if matches!(
                status_module,
                FlowStatusModule::Success { branch_chosen: Some(_), .. }
            ) {
                last_job_result
                    .as_ref()
                    .map(|x| x.as_ref().clone())
                    .unwrap_or_else(|| to_raw_value(&json!("{}")))
            } else {
                // it has to be an empty for loop event
                serde_json::from_str("[]").unwrap()
            },
            stop_early_override: None,
            w_id: flow_job.workspace_id.clone(),
            worker_dir: worker_dir.to_string(),
            token: client.token.clone(),
        })));
    }

    if matches!(step, Step::Step { idx: 0, .. }) {
        if !flow_job.is_flow_step() && flow_job.schedule_path().is_some() {
            let schedule_path = flow_job.schedule_path();
            let no_flow_overlap = sqlx::query_scalar!(
                "SELECT no_flow_overlap FROM schedule WHERE path = $1 AND workspace_id = $2",
                schedule_path.as_ref().unwrap(),
                flow_job.workspace_id.as_str()
            )
            .fetch_one(db)
            .warn_after_seconds(3)
            .await?;
            if no_flow_overlap {
                let overlapping = sqlx::query_scalar!(
                     // Query plan:
                     // - use of the `ix_v2_job_root_by_path` index; hence the `parent_job IS NULL`
                     //   clause.
                     // - select from `v2_job` first, then join with `v2_job_queue` to avoid a full
                     //   table scan on `running = true`.
                     "SELECT id
                     FROM v2_job j JOIN v2_job_queue USING (id)
                     WHERE j.workspace_id = $2 AND trigger_kind = 'schedule' AND trigger = $1 AND runnable_path = $4
                         AND parent_job IS NULL
                         AND j.id != $3
                         AND running = true",
                     schedule_path.as_ref().unwrap(),
                     flow_job.workspace_id.as_str(),
                     flow_job.id,
                     flow_job.runnable_path()
                 )
                 .fetch_all(db)
                 .warn_after_seconds(3)
                 .await?;
                if overlapping.len() > 0 {
                    let overlapping_str = overlapping
                        .iter()
                        .map(|x| x.to_string())
                        .collect::<Vec<String>>()
                        .join(", ");

                    return Ok(PushNextFlowJob::Done(Some(
                        UpdateFlow {
                            flow: flow_job.id,
                            success: true,
                            result: serde_json::from_str(
                                &format!("\"not allowed to overlap with {overlapping_str}, scheduling next iteration\""),
                            )
                            .unwrap(),
                            stop_early_override: Some(true),
                            w_id: flow_job.workspace_id.clone(),
                            worker_dir: worker_dir.to_string(),
                            token: client.token.clone(),
                        }
                    )));
                }
            }
        }
        if let Some(skip_expr) = &flow.skip_expr {
            let skip_res = compute_bool_from_expr(
                &skip_expr,
                arc_flow_job_args.clone(),
                flow_env,
                Arc::new(to_raw_value(&json!("{}"))),
                None,
                None,
                Some(client),
                None,
                Some(vec![(
                    windmill_common::variables::WM_SCHEDULED_FOR.to_string(),
                    flow_job.scheduled_for.to_string(),
                )]),
            )
            .warn_after_seconds(3)
            .await;
            if let Err(e) = &skip_res {
                append_predicate_error_to_root_logs(
                    db,
                    root_flow_id_for(&flow_job),
                    &flow_job.workspace_id,
                    None,
                    "flow skip_expr",
                    skip_expr,
                    e,
                )
                .await;
            }
            let skip = skip_res?;
            if skip {
                return Ok(PushNextFlowJob::Done(Some(UpdateFlow {
                    flow: flow_job.id,
                    success: true,
                    result: serde_json::from_str("\"stopped early\"").unwrap(),
                    stop_early_override: Some(true),
                    w_id: flow_job.workspace_id.clone(),
                    worker_dir: worker_dir.to_string(),
                    token: client.token.clone(),
                })));
            }
        }
    }

    // Compute and initialize last_job_result
    let mut 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 { idx: 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 {
        // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job
        // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a
        // flow restart from a specific step
        if last_job_result.is_some() {
            last_job_result.unwrap()
        } else {
            match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status)
                .warn_after_seconds(3)
                .await?
            {
                None => Arc::new(to_raw_value(&json!("{}"))),
                Some(previous_job_result) => Arc::new(previous_job_result),
            }
        }
    };

    let mut resume_messages: Vec<Box<RawValue>> = vec![];
    let mut approvers: Vec<String> = vec![];

    /* (suspend / resume), when starting a module, if previous module has a
     * non-zero `suspend` value, collect `resume_job`s for the previous module job.
     *
     * If there aren't enough, try again later. */
    if matches!(
        &status_module,
        FlowStatusModule::WaitingForPriorSteps { .. } | FlowStatusModule::WaitingForEvents { .. }
    ) {
        if let Some((suspend, last)) = needs_resume(&flow, &status) {
            let mut tx = db.begin().warn_after_seconds(3).await?;

            /* Lock this row to prevent the suspend column getting out out of sync
             * if a resume message arrives after we fetch and count them here.
             *
             * This only works because jobs::resume_job does the same thing. */
            sqlx::query_scalar!(
                "SELECT null FROM v2_job_queue WHERE id = $1 FOR UPDATE",
                flow_job.id
            )
            .fetch_one(&mut *tx)
            .warn_after_seconds(3)
            .await
            .context("lock flow in queue")?;

            // Query for both step-level resumes (job = step_id) and flow-level resumes (job = flow_id)
            // Flow-level resumes allow pre-approvals that can be consumed by any suspend step
            let resumes = sqlx::query_as::<_, ResumeRow>(
                 "SELECT value, approver, resume_id, approved FROM resume_job WHERE job = $1 OR job = $2 ORDER BY created_at ASC",
             )
             .bind(last)
             .bind(flow_job.id)
             .fetch_all(&mut *tx)
             .warn_after_seconds(3)
             .await
             ?
             .into_iter()
             .collect::<Vec<_>>();

            resume_messages.extend(resumes.iter().map(|r| to_raw_value(&r.value)));
            approvers.extend(resumes.iter().map(|r| {
                r.approver
                    .clone()
                    .as_deref()
                    .unwrap_or_else(|| "anonymous")
                    .to_string()
            }));

            // Persist approval user groups conditions, if any. Requires runnning the InputTransform
            let required_events = suspend.required_events.unwrap() as u16;
            let user_auth_required = suspend.user_auth_required.unwrap_or(false);
            if user_auth_required {
                let self_approval_disabled = suspend.self_approval_disabled.unwrap_or(false);
                let user_groups_required: Vec<String>;
                if let Some(user_groups_required_as_input_transform) = suspend.user_groups_required
                {
                    match user_groups_required_as_input_transform {
                        InputTransform::Static { value } => {
                            user_groups_required = serde_json::from_str::<Vec<String>>(value.get())
                                .expect("Unable to deserialize group names");
                        }
                        InputTransform::Javascript { expr } => {
                            let mut context = HashMap::with_capacity(2);
                            context.insert("result".to_string(), arc_last_job_result.clone());
                            context
                                .insert("previous_result".to_string(), arc_last_job_result.clone());

                            let eval_result = serde_json::from_str::<Vec<String>>(
                                 eval_timeout(
                                     expr.to_string(),
                                     context,
                                     Some(arc_flow_job_args.clone()),
                                     flow_env,
                                     None,
                                     None,
                                     None
                                 )
                                 .warn_after_seconds(3)
                                 .await
                                 .map_err(|e| {
                                     Error::ExecutionErr(format!(
                                         "Error during isolated evaluation of expression `{expr}`:\n{e:#}"
                                     ))
                                 })?
                                 .get(),
                             );
                            if eval_result.is_ok() {
                                user_groups_required = eval_result.ok().unwrap_or(Vec::new())
                            } else {
                                let e = eval_result.err().unwrap();
                                return Err(Error::ExecutionErr(format!(
                                    "Result returned by input transform invalid `{e:#}`"
                                )));
                            }
                        }
                        InputTransform::Ai => {
                            user_groups_required = Vec::new();
                        }
                    }
                } else {
                    user_groups_required = Vec::new();
                };

                let approval_conditions = ApprovalConditions {
                    user_auth_required,
                    user_groups_required,
                    self_approval_disabled,
                };
                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['approval_conditions'], $1)
                     WHERE id = $2",
                    json!(approval_conditions),
                    flow_job.id
                )
                .execute(&mut *tx)
                .warn_after_seconds(3)
                .await?;
            }

            let is_disapproved = resumes.iter().find(|x| !x.approved);
            let can_be_resumed =
                is_disapproved.is_none() && resume_messages.len() >= required_events as usize;
            let disapproved_or_timeout_but_continue = !can_be_resumed
                && (is_disapproved.is_some()
                    || !matches!(
                        &status_module,
                        FlowStatusModule::WaitingForPriorSteps { .. }
                    ))
                && suspend.continue_on_disapprove_timeout.unwrap_or(false);

            let audit_author = AuditAuthor {
                username: flow_job
                    .permissioned_as
                    .trim_start_matches("u/")
                    .to_string(),
                email: flow_job.permissioned_as_email.clone(),
                username_override: None,
                token_prefix: Some(format!("psh.nxt.flowjob-{}", client.token.to_string())),
            };

            if can_be_resumed || disapproved_or_timeout_but_continue {
                if disapproved_or_timeout_but_continue {
                    let js = if let Some(disapproved) = is_disapproved.as_ref() {
                        json!({"error": {"message": format!("Disapproved by {}", disapproved.approver.clone().unwrap_or_else( || "unknown".to_string())), "name": "SuspendedDisapproved"}})
                    } else {
                        json!({"error": {"message": "Timed out waiting to be resumed", "name": "SuspendedTimedOut"}})
                    };

                    resume_messages.push(to_raw_value(&js));
                    audit_log(
                         &mut *tx,
                         &audit_author,
                         "jobs.suspend_resume",
                         ActionKind::Update,
                         &flow_job.workspace_id,
                         Some(&serde_json::json!({"approved": false, "job_id": flow_job.id, "details": "Suspend timed out without approval but can continue".to_string()}).to_string()),
                         None,
                     )
                     .warn_after_seconds(3)
                     .await?;
                }

                sqlx::query!(
                     "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'approvers'], $2)
                     WHERE id = $3",
                     (status.step - 1).to_string(),
                     json!(resumes
                         .into_iter()
                         .map(|r| Approval {
                             resume_id: r.resume_id as u16,
                             approver: r
                                 .approver.clone()
                                 .unwrap_or_else(|| "unknown".to_string())
                         })
                         .collect::<Vec<_>>()
                     ),
                     flow_job.id
                 )
                 .execute(&mut *tx)
                 .warn_after_seconds(3)
                 .await?;

                // Remove the approval conditions from the flow status
                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = flow_status - 'approval_conditions'
                     WHERE id = $1",
                    flow_job.id
                )
                .execute(&mut *tx)
                .warn_after_seconds(3)
                .await?;

                /* continue on and run this job! */
                tx.commit().warn_after_seconds(3).await?;

            /* not enough messages to do this job, "park"/suspend until there are */
            } else if matches!(
                &status_module,
                FlowStatusModule::WaitingForPriorSteps { .. }
            ) && is_disapproved.is_none()
            {
                sqlx::query!(
                    "WITH suspend AS (
                         UPDATE v2_job_queue SET suspend = $2, suspend_until = now() + $3
                         WHERE id = $4
                         RETURNING id
                     ) UPDATE v2_job_status SET flow_status = JSONB_SET(
                         flow_status,
                         ARRAY['modules', flow_status->>'step'::TEXT],
                         $1
                     ) WHERE id = (SELECT id FROM suspend)",
                    json!(FlowStatusModule::WaitingForEvents {
                        id: status_module.id(),
                        count: required_events,
                        job: last
                    }),
                    (required_events - resume_messages.len() as u16) as i32,
                    Duration::from_secs(
                        suspend.timeout.map(|t| t.into()).unwrap_or_else(|| 30 * 60)
                    ) as Duration,
                    flow_job.id,
                )
                .execute(&mut *tx)
                .warn_after_seconds(3)
                .await?;

                sqlx::query!(
                    "UPDATE v2_job_runtime SET ping = NULL
                     WHERE id = $1",
                    flow_job.id,
                )
                .execute(&mut *tx)
                .warn_after_seconds(3)
                .await?;

                tx.commit().warn_after_seconds(3).await?;
                return Ok(PushNextFlowJob::Done(None));

            /* cancelled or we're WaitingForEvents but we don't have enough messages (timed out) */
            } else {
                if is_disapproved.is_none() {
                    audit_log(
                         &mut *tx,
                         &audit_author,
                         "jobs.suspend_resume",
                         ActionKind::Update,
                         &flow_job.workspace_id,
                         Some(&serde_json::json!({"approved": false, "job_id": flow_job.id, "details": "Suspend timed out without approval and is cancelled".to_string()}).to_string()),
                         None,
                     )
                     .warn_after_seconds(3)
                     .await?;
                }
                tx.commit().warn_after_seconds(3).await?;

                let (logs, error_name) = if let Some(disapprover) = is_disapproved {
                    (
                        format!(
                            "Disapproved by {}",
                            disapprover
                                .approver
                                .clone()
                                .unwrap_or_else(|| "unknown".to_string())
                        ),
                        "SuspendedDisapproved",
                    )
                } else {
                    (
                        "Timed out waiting to be resumed".to_string(),
                        "SuspendedTimedOut",
                    )
                };

                let result: Value = json!({ "error": {"message": logs, "name": error_name}});

                append_logs(
                    &flow_job.id,
                    &flow_job.workspace_id,
                    logs.clone(),
                    &db.into(),
                )
                .warn_after_seconds(3)
                .await;

                return Ok(PushNextFlowJob::Done(Some(UpdateFlow {
                    flow: flow_job.id,
                    success: false,
                    result: to_raw_value(&result),
                    stop_early_override: None,
                    w_id: flow_job.workspace_id.clone(),
                    worker_dir: worker_dir.to_string(),
                    token: client.token.clone(),
                })));
            }
        }
    }

    let mut module = flow.get_flow_module_at_step(step)?;

    let current_id = &module.id;
    let mut previous_id = match step {
        Step::Step { idx: i, .. } if i >= 1 => {
            flow.modules.get(i - 1).map(|m| m.id.clone()).unwrap()
        }
        _ => String::new(),
    };

    // calculate sleep if any
    let mut scheduled_for_o = {
        // avoid branchall sleeping on every iteration if sleep is on module prior
        if !matches!(
            &status_module,
            FlowStatusModule::WaitingForPriorSteps { .. }
                | FlowStatusModule::WaitingForEvents { .. }
        ) {
            None
        } else {
            let sleep_input_transform = if let Step::Step { idx: i, .. } = step {
                i.checked_sub(1)
                    .and_then(|i| flow.modules.get(i))
                    .and_then(|m| m.sleep.clone())
            } else {
                None
            };

            if let Some(input_transform) = sleep_input_transform {
                let timeout_value = evaluate_input_transform::<serde_json::Value>(
                    &input_transform,
                    arc_last_job_result.clone(),
                    Some(arc_flow_job_args.clone()),
                    flow_env,
                    Some(client),
                    None,
                )
                .await;
                match timeout_value {
                    Ok(serde_json::Value::Number(n)) => {
                        if n.is_f64() {
                            n.as_f64()
                                .map(|x: f64| from_now(Duration::from_millis((x * 1000.0) as u64)))
                        } else if n.is_u64() {
                            n.as_u64().map(|x: u64| from_now(Duration::from_secs(x)))
                        } else {
                            return Err(Error::ExecutionErr(format!(
                                "Expected an integer, found: {n}"
                            )));
                        }
                    }
                    Ok(x @ _) => Err(Error::ExecutionErr(format!(
                        "Expected an integer, found: {x:?}"
                    )))?,
                    Err(e) => Err(Error::ExecutionErr(format!(
                        "Expected a number value, had error instead: {e:?}",
                    )))?,
                }
            } else {
                None
            }
        }
    };

    let retry = if matches!(&status_module, FlowStatusModule::Failure { .. },) {
        let retry = &module.retry.clone().unwrap_or_default();
        evaluate_retry(
            retry,
            &status.retry,
            arc_last_job_result.clone(),
            arc_flow_job_args.clone(),
            flow_env,
            Some(client),
        )
        .await?
    } else {
        None
    };
    let get_args_from_id = match &status_module {
        FlowStatusModule::Failure { job, .. }
            if retry.as_ref().is_some() || !module.continue_on_error.is_some_and(|x| x) =>
        {
            if let Some((fail_count, retry_in)) = retry {
                tracing::debug!(
                    retry_in_seconds = retry_in.as_secs(),
                    fail_count = fail_count,
                    "retrying"
                );

                scheduled_for_o = Some(from_now(retry_in));
                status.retry.failed_jobs.push(job.clone());
                sqlx::query!(
                     "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(JSONB_SET(flow_status, ARRAY['retry'], $1), ARRAY['modules', $3::TEXT, 'failed_retries'], $4)
                     WHERE id = $2",
                     json!(RetryStatus { fail_count, ..status.retry.clone() }),
                     flow_job.id,
                     status.step.to_string(),
                     json!(status.retry.failed_jobs)
                 )
                 .execute(db)
                 .warn_after_seconds(2)
                 .await
                 .context("update flow retry")?;

                status_module = FlowStatusModule::WaitingForPriorSteps { id: status_module.id() };

                // The failed attempt's error has already been consumed by `evaluate_retry`
                // above. Restore `previous_result` to the preceding step's result (or the
                // flow args for the first step) so that predicates re-evaluated for the
                // retry (skip_if, loop iterator expressions, ...) don't see the failed
                // attempt's error instead. Like the suspend/restart path above, this
                // falls back to `"{}"` when the preceding step has no Success status
                // (e.g. it failed with continue_on_error).
                if !matches!(step, Step::FailureStep) {
                    arc_last_job_result = if matches!(step, Step::Step { idx: 0, .. })
                        || step.is_preprocessor_step()
                    {
                        Arc::new(to_raw_value(&flow_job.args))
                    } else {
                        match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status)
                            .warn_after_seconds(3)
                            .await?
                        {
                            None => Arc::new(to_raw_value(&json!("{}"))),
                            Some(previous_job_result) => Arc::new(previous_job_result),
                        }
                    };
                }

                // we get the args from the last failed job
                status.retry.failed_jobs.last()
            /* Start the failure module ... */
            } else {
                /* push_next_flow_job is called with the current step on FlowStatusModule::Failure.
                 * This must update the step index to the end so that no subsequent steps are run after
                 * the failure module.
                 *
                 * The failure module may also run again if it fails and the retry feature is used.
                 * In that case, `i` will index past `flow.modules`.  The above should handle that and
                 * re-run the failure module. */
                step = Step::FailureStep;
                previous_id = current_id.clone();
                module = flow
                    .failure_module
                    .as_ref()
                    /* If this fails, it's a update_flow_status_after_job_completion shouldn't have called
                     * handle_flow to get here. */
                    .context("missing failure module")?;
                status_module = status.failure_module.module_status.clone();

                if module.retry.as_ref().is_some_and(|x| x.has_attempts()) {
                    sqlx::query!(
                        "UPDATE v2_job_status
                         SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1)
                         WHERE id = $2",
                        json!(RetryStatus { fail_count: 0, failed_jobs: vec![] }),
                        flow_job.id
                    )
                    .execute(db)
                    .warn_after_seconds(3)
                    .await
                    .context("update flow retry")?;
                };
                None
            }
        }
        _ => None,
    };

    let mut transform_context: Option<IdContext> = None;

    let approvers = Arc::new(to_raw_value(&approvers));
    let resume = Arc::new(to_raw_value(&resume_messages.last()));
    let resumes = Arc::new(to_raw_value(&resume_messages));

    drop(resume_messages);

    let is_skipped = if let Some(skip_if) = &module.skip_if {
        let idcontext = get_transform_context(&flow_job, previous_id.as_str(), &status);
        let skip_if_res = compute_bool_from_expr(
            &skip_if.expr,
            arc_flow_job_args.clone(),
            flow_env,
            arc_last_job_result.clone(),
            None,
            Some(&idcontext),
            Some(client),
            Some((resumes.clone(), resume.clone(), approvers.clone())),
            None,
        )
        .warn_after_seconds(3)
        .await;
        if let Err(e) = &skip_if_res {
            append_predicate_error_to_root_logs(
                db,
                root_flow_id_for(&flow_job),
                &flow_job.workspace_id,
                Some(&module.id),
                "skip_if",
                &skip_if.expr,
                e,
            )
            .await;
        }
        skip_if_res?
    } else {
        false
    };

    let args: windmill_common::error::Result<_> = if module.mock.is_some()
        && module.mock.as_ref().unwrap().enabled
    {
        let mut hm = HashMap::new();
        hm.insert(
            "previous_result".to_string(),
            to_raw_value(
                &module
                    .mock
                    .as_ref()
                    .unwrap()
                    .return_value
                    .clone()
                    .unwrap_or_else(|| serde_json::from_str("null").unwrap()),
            ),
        );
        Ok(Marc::new(hm))
    } else if let Some(id) = get_args_from_id {
        let args = sqlx::query_scalar!(
            "SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"
                 FROM v2_job WHERE id = $1 AND workspace_id = $2",
            id,
            &flow_job.workspace_id
        )
        .fetch_optional(db)
        .warn_after_seconds(3)
        .await?;
        if let Some(args) = args {
            Ok(Marc::new(args.map(|x| x.0).unwrap_or_else(HashMap::new)))
        } else {
            Ok(Marc::new(HashMap::new()))
        }
    } else if step.is_preprocessor_step() {
        let mut hm = (*arc_flow_job_args).clone();
        hm.insert(
            ENTRYPOINT_OVERRIDE.to_string(),
            to_raw_value(&"preprocessor"),
        );
        Ok(Marc::new(hm))
    } else if module.pass_flow_input_directly.unwrap_or(false) {
        // If pass_flow_input_directly is set, use flow args directly
        Ok(arc_flow_job_args.clone())
    } else {
        let value = module.get_value();
        match &value {
            Ok(_) if matches!(value, Ok(FlowModuleValue::Identity)) || is_skipped => {
                serde_json::from_str(
                    &serde_json::to_string(&PreviousResult {
                        previous_result: Some(&arc_last_job_result),
                    })
                    .unwrap(),
                )
                .map(Marc::new)
                .map_err(|e| error::Error::internal_err(format!("identity: {e:#}")))
            }
            Ok(
                FlowModuleValue::Script { input_transforms, .. }
                | FlowModuleValue::RawScript { input_transforms, .. }
                | FlowModuleValue::FlowScript { input_transforms, .. }
                | FlowModuleValue::Flow { input_transforms, .. }
                | FlowModuleValue::AIAgent { input_transforms, .. },
            ) => {
                let ctx = get_transform_context(&flow_job, &previous_id, &status);
                transform_context = Some(ctx);
                let by_id = transform_context.as_ref().unwrap();
                // if a failure step, we add flow job id and started_at to the context. This is for error handling of triggers where we wrap scripts into single step flows
                // It's to make its arguments consistent with global/workspace/schedule error handlers.
                let failure_context = match step {
                    Step::FailureStep if flow_job.started_at.is_some() => Some(FailureContext {
                        started_at: Arc::new(to_raw_value(&flow_job.started_at.unwrap())),
                        flow_job_id: flow_job.id,
                    }),
                    _ => None,
                };
                transform_input(
                    arc_flow_job_args.clone(),
                    flow_env,
                    arc_last_job_result.clone(),
                    input_transforms,
                    resumes.clone(),
                    resume.clone(),
                    approvers.clone(),
                    failure_context,
                    by_id,
                    client,
                )
                .warn_after_seconds(3)
                .await
                .map(Marc::new)
            }
            Ok(_) => Ok(arc_flow_job_args.clone()),
            Err(e) => {
                return Err(error::Error::internal_err(format!(
                    "module was not convertible to acceptable value {e:?}"
                )))
            }
        }
    };
    tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed");

    let next_flow_transform = compute_next_flow_transform(
        arc_flow_job_args.clone(),
        arc_last_job_result.clone(),
        flow_env,
        &flow_job,
        &flow,
        transform_context,
        db,
        &module,
        &status,
        &status_module,
        &previous_id,
        client,
        resumes.clone(),
        resume.clone(),
        approvers.clone(),
        is_skipped,
    )
    .warn_after_seconds(3)
    .await?;
    tracing::info!(id = %flow_job.id, root_id = %job_root, "next flow transform computed");

    let (job_payloads, next_status) = match next_flow_transform {
        NextFlowTransform::Continue(job_payload, next_state) => (job_payload, next_state),
        NextFlowTransform::EmptyInnerFlows { branch_chosen } => {
            let raw_status = sqlx::query_scalar!(
                "UPDATE v2_job_status
                 SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)
                 WHERE id = $3
                 RETURNING flow_status AS \"flow_status: Json<Box<RawValue>>\"",
                status.step.to_string(),
                json!(FlowStatusModule::Success {
                    id: status_module.id(),
                    job: Uuid::nil(),
                    flow_jobs: if branch_chosen.is_some() {
                        None
                    } else {
                        Some(vec![])
                    },
                    flow_jobs_success: if branch_chosen.is_some() {
                        None
                    } else {
                        Some(vec![])
                    },
                    flow_jobs_duration: if branch_chosen.is_some() {
                        None
                    } else {
                        Some(FlowJobsDuration { started_at: vec![], duration_ms: vec![] })
                    },
                    branch_chosen: branch_chosen,
                    approvers: vec![],
                    failed_retries: vec![],
                    skipped: false,
                    agent_actions: None,
                    agent_actions_success: None,
                }),
                flow_job.id
            )
            .fetch_optional(db)
            .warn_after_seconds(3)
            .await?
            .flatten();

            let status = raw_status
                .as_ref()
                .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok());

            if let Some(status) = status {
                // // flow is reprocessed by the worker in a state where the module has completed successfully.
                return Ok(PushNextFlowJob::Rec(PushNextFlowJobRec {
                    flow_job: flow_job,
                    status: status,
                }));
            } else {
                return Err(Error::BadRequest(
                    "impossible to parse new flow status after applying inner flows".to_string(),
                ));
            }
        }
        NextFlowTransform::StepFailure { error } => {
            let result = Arc::new(to_raw_value(&WrappedError { error }));
            update_flow_status_after_job_completion(
                db,
                client,
                flow_job.id,
                &Uuid::nil(),
                &flow_job.workspace_id,
                false,
                None,
                result,
                None,
                false,
                same_worker_tx,
                worker_dir,
                None,
                worker_name,
                job_completed_tx,
                flow_runners,
                killpill_rx,
                #[cfg(feature = "benchmark")]
                &mut BenchmarkIter::new(),
            )
            .await?;
            return Ok(PushNextFlowJob::Done(None));
        }
    };

    // only start runners if we're not already in a squash for loop
    let start_runners = flow_runners.is_none()
        && matches!(
            next_status,
            NextStatus::NextLoopIteration { start_runners: true, .. }
        );

    let do_not_pass_runners = matches!(next_status, NextStatus::NextStep { .. })
        && flow_runners
            .as_ref()
            .is_some_and(|fr| fr.job_id == flow_job.id);

    let continue_with_runners = (start_runners || (flow_runners.is_some() && !do_not_pass_runners))
        && module.suspend.is_none()
        && module.sleep.is_none();

    // Also check `flow_job.same_worker` for [`JobKind::Flow`] jobs as it's no
    // more reflected to the flow value on push.
    let job_same_worker = flow_job.same_worker
        && matches!(flow_job.kind, JobKind::Flow)
        && flow_job.runnable_id.is_some();
    let continue_on_same_worker =
        (flow.same_worker || job_same_worker) && module.suspend.is_none() && module.sleep.is_none();

    /* Finally, push the job into the queue */
    let mut uuids = vec![];

    let job_payloads = match job_payloads {
        ContinuePayload::SingleJob(payload) => vec![payload],
        ContinuePayload::ParallelJobs(payloads) => payloads,
    };
    let len = job_payloads.len();

    let mut tx = db.begin().warn_after_seconds(3).await?;
    let nargs = args.as_ref();
    for (i, payload_tag) in job_payloads.into_iter().enumerate() {
        if i % 100 == 0 && i != 0 {
            tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}");
            // Ping on the pool, outside `tx`, so the zombie flow monitor sees it before the
            // push transaction commits — otherwise large parallel pushes can be flagged as
            // zombie and trigger a cancel/push deadlock.
            sqlx::query!(
                "UPDATE v2_job_runtime SET ping = now() WHERE id = $1",
                flow_job.id,
            )
            .execute(db)
            .warn_after_seconds(3)
            .await?;
        }
        tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushing job {i} of {len}");

        // compute job-to-be-pushed priority
        // The job definition itself might have its own priority, but as we're running
        // it from a flow here, it inherits first the flow step priority and second the
        // flow priority.
        let new_job_priority_override = if module.priority.is_some() {
            module.priority
        } else if flow_job.priority.is_some() {
            flow_job.priority
        } else {
            None
        };

        let marc;
        let me;
        let args = match &next_status {
            NextStatus::AllFlowJobs {
                branchall: Some(BranchAllStatus { .. }),
                iterator: None,
                ..
            } => nargs,
            NextStatus::NextLoopIteration {
                next: ForloopNextIteration { new_args, .. },
                simple_input_transforms,
                ..
            } => {
                let mut args = if let Ok(args) = nargs {
                    args.as_ref().clone()
                } else {
                    HashMap::new()
                };
                insert_iter_arg(&mut args, "iter".to_string(), to_raw_value(new_args));

                args.insert("iter".to_string(), to_raw_value(new_args));
                if let Some(input_transforms) = simple_input_transforms {
                    //previous id is none because we do not want to use previous id if we are in a for loop
                    let ctx = get_transform_context(&flow_job, "", &status);
                    let ti = transform_input(
                        Marc::new(args),
                        flow_env,
                        arc_last_job_result.clone(),
                        input_transforms,
                        resumes.clone(),
                        resume.clone(),
                        approvers.clone(),
                        None,
                        &ctx,
                        client,
                    )
                    .warn_after_seconds(3)
                    .await
                    .map_err(|e| {
                        Error::ExecutionErr(
                            format!("could not transform input using an expr: {e}",),
                        )
                    })
                    .map(Marc::new);
                    if let Ok(ti) = ti {
                        marc = ti;
                        Ok(&marc)
                    } else {
                        me = ti.unwrap_err();
                        Err(&me)
                    }
                } else {
                    marc = Marc::new(args);
                    Ok(&marc)
                }
            }
            NextStatus::AllFlowJobs {
                branchall: None,
                iterator: Some(FlowIterator { itered, .. }),
                simple_input_transforms,
            } => {
                if let Ok(args) = args.as_ref() {
                    let Some(itered) = itered else {
                        return Err(Error::ExecutionErr(format!(
                            "iterator itered should never be none for parallel for loop jobs"
                        )));
                    };

                    let mut hm = HashMap::new();
                    for (k, v) in args.iter() {
                        hm.insert(k.to_string(), v.to_owned());
                    }
                    insert_iter_arg(
                        &mut hm,
                        "iter".to_string(),
                        to_raw_value(&json!({ "index": i as i32, "value": itered[i]})),
                    );
                    if let Some(input_transforms) = simple_input_transforms {
                        let ctx = get_transform_context(&flow_job, &previous_id, &status);
                        let ti = transform_input(
                            Marc::new(hm),
                            flow_env,
                            arc_last_job_result.clone(),
                            input_transforms,
                            resumes.clone(),
                            resume.clone(),
                            approvers.clone(),
                            None,
                            &ctx,
                            client,
                        )
                        .warn_after_seconds(3)
                        .await
                        .map_err(|e| {
                            Error::ExecutionErr(format!(
                                "could not transform input using an expr: {e}"
                            ))
                        })
                        .map(Marc::new);
                        if let Ok(ti) = ti {
                            marc = ti;
                            Ok(&marc)
                        } else {
                            me = ti.unwrap_err();
                            Err(&me)
                        }
                    } else {
                        marc = Marc::new(hm);
                        Ok(&marc)
                    }
                } else {
                    nargs
                }
            }
            _ => nargs,
        };

        let mut push_args;
        let err;
        let ov;

        match args {
            Ok(v) => {
                ov = v;
                push_args = PushArgs::from(ov.as_ref());
                err = None;
            }
            Err(e) => {
                push_args = PushArgs::from(&*EHM);
                err = Some(e);
            }
        };

        // Propagate temp script refs from the flow preview job to each step so
        // relative imports in inline scripts resolve from not-yet-deployed local
        // content (uploaded to raw_script_temp) instead of the deployed script.
        // Gated on JobKind::FlowPreview because flow_job.args includes
        // caller-controlled request args; honoring this key on deployed flow
        // runs would let a caller swap import resolution targets in deployed
        // step code.
        if matches!(flow_job.kind, JobKind::FlowPreview) {
            if let Some(temp_script_refs) = arc_flow_job_args.as_ref().get("_TEMP_SCRIPT_REFS") {
                push_args
                    .extra
                    .get_or_insert_with(HashMap::new)
                    .insert("_TEMP_SCRIPT_REFS".to_string(), temp_script_refs.clone());
            }
        }

        // Propagate the inbound W3C traceparent captured at enqueue to each step
        // so the whole flow shares the originating distributed trace (the trace
        // identity is otherwise derived from the root job UUID). Observability
        // only — no security impact — so unlike _TEMP_SCRIPT_REFS it is not
        // gated to previews.
        if let Some(traceparent) = arc_flow_job_args
            .as_ref()
            .get(windmill_common::jobs::WM_TRACEPARENT)
        {
            push_args.extra.get_or_insert_with(HashMap::new).insert(
                windmill_common::jobs::WM_TRACEPARENT.to_string(),
                traceparent.clone(),
            );
        }

        tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed args for job {i} of {len}");

        let value_with_parallel = module.get_value_with_parallel()?;

        let flow_innermost_root_job = if {
            value_with_parallel.type_ == "flow"
                || (value_with_parallel.type_ == "forloopflow"
                    && value_with_parallel.parallel.is_some_and(|x| x))
        } {
            None
        } else {
            flow_job
                .flow_innermost_root_job
                .or_else(|| Some(flow_job.id))
        };

        let flow_root_job = get_root_job_id(&flow_job);

        // forward root job permissions to the new job
        let job_perms: Option<Authed> =
            get_job_perms(&mut *tx, &flow_root_job, &flow_job.workspace_id)
                .await?
                .map(|x| x.into());

        tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
        let tag = resolve_flow_step_tag(
            step.is_preprocessor_step(),
            &flow_job.tag,
            &flow_job.workspace_id,
            flow.preserve_step_tags,
            payload_tag.tag.as_deref(),
        );

        let (email, permissioned_as) = if let Some(on_behalf_of) = payload_tag.on_behalf_of.as_ref()
        {
            (&on_behalf_of.email, on_behalf_of.permissioned_as.clone())
        } else {
            (
                &flow_job.permissioned_as_email,
                flow_job.permissioned_as.to_owned(),
            )
        };

        // Check tag availability for flow substeps to prevent abuse.
        // Skip when the substep inherits the parent flow's tag: that tag was already
        // validated at flow push time, and for dedicated flows it is the auto-generated
        // `{workspace_id}:flow/{path}` tag which is never in CUSTOM_TAGS.
        if let Some(tag_str) = tag
            .as_deref()
            .filter(|t| !t.is_empty() && *t != flow_job.tag.as_str())
        {
            check_tag_available_for_workspace_internal(
                &db,
                &flow_job.workspace_id,
                tag_str,
                email,
                None, // no token for flow substeps so no scopes so no scope_tags
            )
            .warn_after_seconds_with_sql(
                1,
                "check_tag_available_for_workspace_internal".to_string(),
            )
            .await?;
        }

        let evaluated_timeout = if let Some(timeout_transform) = &module.timeout {
            let ctx = get_transform_context(&flow_job, &previous_id, &status);

            let timeout_value = evaluate_input_transform::<i32>(
                timeout_transform,
                arc_last_job_result.clone(),
                Some(arc_flow_job_args.clone()),
                flow_env,
                Some(client),
                Some(&ctx),
            )
            .await?;

            if timeout_value < 0 {
                return Err(Error::ExecutionErr(
                    "Timeout value cannot be negative".to_string(),
                ));
            }

            Some(timeout_value)
        } else {
            payload_tag.timeout
        };

        let tx2 = PushIsolationLevel::Transaction(tx);
        let (uuid, mut inner_tx) = push(
            &db,
            tx2,
            &flow_job.workspace_id,
            payload_tag.payload.clone(),
            push_args,
            &flow_job.created_by,
            email,
            permissioned_as,
            Some(&format!(
                "job-span-{}",
                flow_job.flow_innermost_root_job.unwrap_or(flow_job.id)
            )),
            scheduled_for_o,
            flow_job.schedule_path(),
            Some(flow_job.id),
            Some(flow_root_job),
            flow_innermost_root_job,
            None,
            true,
            continue_on_same_worker,
            err,
            flow_job.visible_to_owner,
            tag,
            evaluated_timeout,
            Some(module.id.clone()),
            new_job_priority_override,
            job_perms.as_ref(),
            continue_with_runners,
            None,
            None,
            None,
        )
        .warn_after_seconds(2)
        .await?;

        if continue_on_same_worker || continue_with_runners {
            let _ = sqlx::query!(
                "UPDATE v2_job_queue SET worker = $2 WHERE id = $1",
                uuid,
                worker_name
            )
            .execute(&mut *inner_tx)
            .warn_after_seconds(3)
            .await;
        }

        tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}");

        // Apply flow node debouncing if configured. Skip parallel steps (for-loops, branchall)
        // where len > 1 — debouncing those would cancel sibling sub-jobs within the same run.
        #[cfg(feature = "private")]
        if len == 1 {
            if let Some(ref debouncing) = module.debouncing {
                if debouncing.debounce_delay_s.is_some_and(|d| d > 0) {
                    let debounce_args = if let Ok(ref v) = nargs {
                        windmill_queue::PushArgs::from(v.as_ref())
                    } else {
                        windmill_queue::PushArgs::from(&*EHM)
                    };
                    let flow_path = flow_job.runnable_path().to_string();
                    windmill_queue::jobs_ee::maybe_debounce_flow_node(
                        debouncing,
                        uuid,
                        flow_job.id,
                        &flow_path,
                        &module.id,
                        &flow_job.workspace_id,
                        &debounce_args,
                        &mut inner_tx,
                        &db,
                    )
                    .await?;
                }
            }
        }

        if value_with_parallel.type_ == "forloopflow"
            && value_with_parallel.parallel.unwrap_or(false)
        {
            if let Some(parallelism_transform) = &value_with_parallel.parallelism {
                tracing::debug!(id = %flow_job.id, root_id = %job_root, "evaluating parallelism expression for forloopflow job {uuid}");

                let ctx = get_transform_context(&flow_job, &previous_id, &status);

                let evaluated_parallelism = evaluate_input_transform::<u16>(
                    parallelism_transform,
                    arc_last_job_result.clone(),
                    Some(arc_flow_job_args.clone()),
                    flow_env,
                    Some(client),
                    Some(&ctx),
                )
                .await?;

                tracing::debug!(id = %flow_job.id, root_id = %job_root, "updating suspend for forloopflow job {uuid} with parallelism {evaluated_parallelism}");

                if i as u16 >= evaluated_parallelism {
                    sqlx::query!(
                        "UPDATE v2_job_queue SET
                             suspend = $1,
                             suspend_until = now() + interval '14 day',
                             running = true
                         WHERE id = $2",
                        (i as u16 - evaluated_parallelism + 1) as i32,
                        uuid,
                    )
                    .execute(&mut *inner_tx)
                    .warn_after_seconds(3)
                    .await?;
                }
                tracing::debug!(id = %flow_job.id, root_id = %job_root, "updated suspend for {uuid}");
            }
        }

        {
            use windmill_common::jobs::resolve_delete_after_secs;
            let resolved = resolve_delete_after_secs(
                payload_tag.delete_after_use.then_some(true),
                payload_tag.delete_after_secs,
            );
            let root_id = flow_innermost_root_job.unwrap_or(flow_job.id);
            match resolved {
                Some(0) => {
                    // Immediate deletion: track in flow_jobs_to_clean (existing behavior)
                    let uuid_singleton_json = serde_json::to_value(&[uuid]).map_err(|e| {
                        error::Error::internal_err(format!("Unable to serialize uuid: {e:#}"))
                    })?;
                    sqlx::query!(
                        "UPDATE v2_job_status
                         SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)
                         WHERE id = $2",
                        uuid_singleton_json,
                        root_id
                    )
                    .execute(&mut *inner_tx)
                    .warn_after_seconds(3)
                    .await?;
                }
                Some(secs) => {
                    // Scheduled deletion: track in flow_jobs_to_schedule_clean
                    let entry = windmill_types::flow_status::FlowJobScheduledClean {
                        id: uuid,
                        delete_after_secs: secs,
                    };
                    let entry_json = serde_json::to_value(&[entry]).map_err(|e| {
                        error::Error::internal_err(format!(
                            "Unable to serialize scheduled clean entry: {e:#}"
                        ))
                    })?;
                    sqlx::query!(
                        "UPDATE v2_job_status
                         SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)
                         WHERE id = $2",
                        entry_json,
                        root_id
                    )
                    .execute(&mut *inner_tx)
                    .warn_after_seconds(3)
                    .await?;
                }
                None => {}
            }
        }

        tx = inner_tx;
        uuids.push(uuid);
    }

    let is_one_uuid = uuids.len() == 1;

    let one_uuid = if is_one_uuid {
        Ok(uuids[0].clone())
    } else {
        Err(Error::BadRequest("Expected only one uuid".to_string()))
    };

    if !is_one_uuid {
        for uuid in &uuids {
            sqlx::query!(
                "INSERT INTO parallel_monitor_lock (parent_flow_id, job_id)
                 VALUES ($1, $2)",
                flow_job.id,
                uuid
            )
            .execute(&mut *tx)
            .warn_after_seconds(3)
            .await?;
            tracing::debug!(id = %flow_job.id, root_id = %job_root, "updated parallel monitor lock for {uuid}");
        }
    }

    let first_uuid = uuids[0];
    let new_status = match next_status {
        NextStatus::NextLoopIteration {
            next:
                ForloopNextIteration {
                    index,
                    itered,
                    itered_len,
                    mut flow_jobs,
                    while_loop,
                    mut flow_jobs_success,
                    mut flow_jobs_duration,
                    ..
                },
            ..
        } => {
            let uuid = one_uuid?;

            flow_jobs.push(uuid);

            if let Some(flow_jobs_success) = &mut flow_jobs_success {
                flow_jobs_success.push(None);
            }
            if let Some(flow_jobs_duration) = &mut flow_jobs_duration {
                flow_jobs_duration.push(&None);
            }

            // Conditionally write itered to separate table if all workers support it
            // Returns None if written to table, Some(itered) if should be in JSONB
            let itered_for_status = write_itered_to_db(db, flow_job.id, &itered).await?;

            FlowStatusModule::InProgress {
                job: uuid,
                iterator: Some(FlowIterator {
                    index,
                    itered: itered_for_status,
                    itered_len: Some(itered_len),
                }),
                flow_jobs: Some(flow_jobs),
                flow_jobs_success,
                flow_jobs_duration,
                branch_chosen: None,
                branchall: None,
                id: status_module.id(),
                parallel: false,
                while_loop,
                progress: None,
                agent_actions: None,
                agent_actions_success: None,
            }
        }
        NextStatus::AllFlowJobs { iterator, branchall, .. } => {
            let iterator_for_status = if let Some(mut iter) = iterator {
                iter.itered = None; // in parallel for loops, itered is only useful when starting the jobs, no need to store it in status/or the dedicated table
                Some(iter)
            } else {
                None
            };

            FlowStatusModule::InProgress {
                job: flow_job.id,
                iterator: iterator_for_status,
                flow_jobs_success: Some(vec![None; uuids.len()]),
                flow_jobs: Some(uuids.clone()),
                flow_jobs_duration: Some(FlowJobsDuration::new(uuids.len())),
                branch_chosen: None,
                branchall,
                id: status_module.id(),
                parallel: true,
                while_loop: false,
                progress: None,
                agent_actions: None,
                agent_actions_success: None,
            }
        }
        NextStatus::NextBranchStep(NextBranch {
            mut flow_jobs,
            status,
            mut flow_jobs_success,
            mut flow_jobs_duration,
            ..
        }) => {
            let uuid = one_uuid?;
            flow_jobs.push(uuid);
            if let Some(flow_jobs_success) = &mut flow_jobs_success {
                flow_jobs_success.push(None);
            }
            if let Some(flow_jobs_duration) = &mut flow_jobs_duration {
                flow_jobs_duration.push(&None);
            }
            FlowStatusModule::InProgress {
                job: uuid,
                iterator: None,
                flow_jobs: Some(flow_jobs),
                flow_jobs_success,
                flow_jobs_duration,
                branch_chosen: None,
                branchall: Some(status),
                id: status_module.id(),
                parallel: false,
                while_loop: false,
                progress: None,
                agent_actions: None,
                agent_actions_success: None,
            }
        }

        NextStatus::BranchChosen(branch) => FlowStatusModule::InProgress {
            job: one_uuid?,
            iterator: None,
            flow_jobs: None,
            flow_jobs_success: None,
            flow_jobs_duration: None,
            branch_chosen: Some(branch),
            branchall: None,
            id: status_module.id(),
            parallel: false,
            while_loop: false,
            progress: None,
            agent_actions: None,
            agent_actions_success: None,
        },
        NextStatus::NextStep => {
            FlowStatusModule::WaitingForExecutor { id: status_module.id(), job: one_uuid? }
        }
    };

    tracing::debug!("STATUS STEP: {:?} {step:?} {:#?}", status.step, new_status);

    match step {
        Step::FailureStep => {
            sqlx::query!(
                "UPDATE v2_job_status SET
                     flow_status = JSONB_SET(
                         JSONB_SET(flow_status, ARRAY['failure_module'], $1),
                         ARRAY['step'],
                         $2
                     )
                 WHERE id = $3",
                json!(FlowStatusModuleWParent {
                    parent_module: Some(current_id.clone()),
                    module_status: new_status
                }),
                json!(flow.modules.len()),
                flow_job.id
            )
            .execute(&mut *tx)
            .warn_after_seconds(3)
            .await?;
        }
        Step::PreprocessorStep => {
            sqlx::query!(
                "UPDATE v2_job_status SET
                     flow_status = JSONB_SET(
                         JSONB_SET(flow_status, ARRAY['preprocessor_module'], $1),
                         ARRAY['step'],
                         $2
                     )
                 WHERE id = $3",
                json!(new_status),
                json!(-1),
                flow_job.id
            )
            .execute(&mut *tx)
            .warn_after_seconds(3)
            .await?;
        }
        Step::Step { idx: i, .. } => {
            sqlx::query!(
                "UPDATE v2_job_status SET
                     flow_status = JSONB_SET(
                         JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
                         ARRAY['step'],
                         $3
                     )
                 WHERE id = $4",
                i as i32,
                json!(new_status),
                json!(i),
                flow_job.id
            )
            .execute(&mut *tx)
            .warn_after_seconds(3)
            .await?;
        }
    };

    potentially_crash_for_testing();

    sqlx::query!(
        "UPDATE v2_job_runtime SET ping = null WHERE id = $1",
        flow_job.id
    )
    .execute(&mut *tx)
    .warn_after_seconds(3)
    .await?;

    if continue_on_same_worker || continue_with_runners {
        if !is_one_uuid {
            return Err(Error::BadRequest(
                 "Cannot continue on same worker with multiple jobs, parallel cannot be used in conjunction with same_worker".to_string(),
             ));
        }
    }
    tx.commit().warn_after_seconds(3).await?;
    tracing::info!(id = %flow_job.id, root_id = %job_root, "all next flow jobs pushed: {uuids:?}");

    if continue_on_same_worker || continue_with_runners {
        let flow_runners = if start_runners {
            tracing::info!(id = %flow_job.id, "starting flow runners for module {}", module.id);
            let (new_flow_runners, new_flow_runner_handles) =
                crate::dedicated_worker_oss::spawn_flow_module_runners(
                    &flow_job,
                    module,
                    flow.failure_module.as_ref(),
                    &killpill_rx,
                    db,
                    worker_dir,
                    &client.base_internal_url,
                    worker_name,
                    &job_completed_tx,
                )
                .await
                .with_context(|| {
                    format!(
                        "failed to spawn flow module runners for job: {}",
                        flow_job.id
                    )
                })?;

            let flow_runners = FlowRunners {
                runners: new_flow_runners,
                handles: new_flow_runner_handles,
                job_id: flow_job.id,
            };
            Some(Arc::new(flow_runners))
        } else if !do_not_pass_runners {
            flow_runners
        } else {
            None
        };

        same_worker_tx
            .send(SameWorkerPayload { job_id: first_uuid, recoverable: true, flow_runners })
            .warn_after_seconds(3)
            .await
            .map_err(to_anyhow)?;
    }
    return Ok(PushNextFlowJob::Done(None));
}

// async fn jump_to_next_step(
//     status_step: i32,
//     i: usize,
//     job_id: &Uuid,
//     flow: FlowValue,
//     db: &DB,
//     client: &windmill_api_client::Client,
//     status_module: FlowStatusModule,
//     last_result: serde_json::Value,
//     same_worker_tx: Sender<Uuid>,
//     base_internal_url: &str,
// ) -> error::Result<()> {
//     let mut tx = db.begin().await?;

//     let next_step = i
//         .checked_add(1)
//         .filter(|i| (..flow.modules.len()).contains(i));

//     let new_job = sqlx::query_as::<_, QueuedJob>(
//         r#"
//                 UPDATE queue
//                     SET flow_status = JSONB_SET(
//                                       JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
//                                                                ARRAY['step'], $3)
//                     WHERE id = $4
//                 RETURNING *
//                 "#,
//     )
//     .bind(status_step)
//     .bind(json!(status_module))
//     .bind(json!(next_step.unwrap_or(i)))
//     .bind(job_id)
//     .fetch_one(&mut *tx)
//     .await?;

//     tx.commit().await?;

//     let new_status = new_job.parse_flow_status().ok_or_else(|| {
//         Error::ExecutionErr("Impossible to parse new status after jump".to_string())
//     })?;

//     if next_step.is_some() {
//         tracing::debug!("Jumping to next step with flow {flow:#?}");
//         return push_next_flow_job(
//             &new_job,
//             new_status,
//             flow,
//             db,
//             client,
//             last_result,
//             same_worker_tx,
//             base_internal_url,
//         )
//         .await;
//     } else {
//         let success = true;
//         let skipped = false;
//         let logs = "Forloop completed without iteration".to_string();
//         let _uuid =
//             add_completed_job(db, client, &new_job, success, skipped, json!([]), logs).await?;
//         return Ok(());
//     }
// }

/// Some state about the current/last forloop FlowStatusModule used to initialized the next
/// iteration's FlowStatusModule after pushing a job
#[derive(Debug)]
struct ForloopNextIteration {
    index: usize,
    itered: Vec<Box<RawValue>>,
    itered_len: usize,
    flow_jobs: Vec<Uuid>,
    flow_jobs_success: Option<Vec<Option<bool>>>,
    flow_jobs_duration: Option<FlowJobsDuration>,
    new_args: Iter,
    while_loop: bool,
}

enum ForLoopStatus {
    ParallelIteration { itered: Vec<Box<RawValue>>, itered_len: usize },
    NextIteration(ForloopNextIteration),
    EmptyIterator,
}

#[derive(Debug)]
struct NextBranch {
    status: BranchAllStatus,
    flow_jobs: Vec<Uuid>,
    flow_jobs_success: Option<Vec<Option<bool>>>,
    flow_jobs_duration: Option<FlowJobsDuration>,
}

#[derive(Debug)]
enum NextStatus {
    NextStep,
    BranchChosen(BranchChosen),
    NextBranchStep(NextBranch),
    NextLoopIteration {
        next: ForloopNextIteration,
        simple_input_transforms: Option<HashMap<String, InputTransform>>,
        start_runners: bool,
    },
    AllFlowJobs {
        branchall: Option<BranchAllStatus>,
        iterator: Option<FlowIterator>,
        simple_input_transforms: Option<HashMap<String, InputTransform>>,
    },
}

#[derive(Debug, Clone)]
pub struct JobPayloadWithTag {
    pub payload: JobPayload,
    pub tag: Option<String>,
    pub delete_after_use: bool,
    pub delete_after_secs: Option<i32>,
    pub timeout: Option<i32>,
    pub on_behalf_of: Option<OnBehalfOf>,
}
#[derive(Debug)]
enum ContinuePayload {
    SingleJob(JobPayloadWithTag),
    ParallelJobs(Vec<JobPayloadWithTag>),
}

#[derive(Debug)]
enum NextFlowTransform {
    EmptyInnerFlows { branch_chosen: Option<BranchChosen> },
    Continue(ContinuePayload, NextStatus),
    // The current module failed in-place (e.g. a BranchOne predicate threw).
    // The error is reported as the step's result so the normal flow machinery
    // (including `failure_module` and surrounding `skip_failures`) applies.
    StepFailure { error: serde_json::Value },
}

fn insert_iter_arg(
    args: &mut HashMap<String, Box<RawValue>>,
    desired_key: String,
    value: Box<RawValue>,
) {
    let conflicting_parent_maybe = args.get(desired_key.as_str());
    if let Some(conflicting_parent) = conflicting_parent_maybe {
        // we change the key of the parent to {desired_key}_parent.
        // we do it recursively b/c it's possible there's another conflict
        let conflicting_parent_new_key = format!("{}_parent", desired_key.as_str());
        insert_iter_arg(
            args,
            conflicting_parent_new_key,
            conflicting_parent.to_owned(),
        );
    }
    args.insert(desired_key, value);
}

/// If the parent flow is a restart whose `nested` chain applies to the child about
/// to be spawned for `module_id` at iteration/branch `iter_index` (None for non-loop
/// containers), returns a `RestartedFlow` payload to use instead of the freshly-built
/// one. The semantics:
/// - `restarted_from.step_id` must equal `module_id`
/// - For loop/branchall containers, `restarted_from.branch_or_iteration_n` is the
///   0-based index of the iteration to restart at, and must equal `iter_index`
/// - For BranchOne / Subflow (single-child), `iter_index` is None and
///   `branch_or_iteration_n` is None or 0
fn nested_restart_payload(
    flow_status: &FlowStatus,
    module_id: &str,
    iter_index: Option<usize>,
) -> Option<JobPayload> {
    let restarted_from = flow_status.restarted_from.as_ref()?;
    if restarted_from.step_id != module_id {
        return None;
    }
    let n = restarted_from.branch_or_iteration_n.unwrap_or(0);
    let matches = match iter_index {
        Some(i) => i == n,
        None => n == 0,
    };
    if !matches {
        return None;
    }
    let nested = restarted_from.nested.as_ref()?;
    let nested = (**nested).clone();
    Some(JobPayload::RestartedFlow {
        completed_job_id: nested.flow_job_id,
        step_id: nested.step_id,
        branch_or_iteration_n: nested.branch_or_iteration_n,
        flow_version: nested.flow_version,
        branch_chosen: nested.branch_chosen,
        nested: nested.nested,
    })
}

fn payload_from_modules<'a>(
    mut modules: Vec<FlowModule>,
    modules_node: Option<FlowNodeId>,
    failure_module: Option<&Box<FlowModule>>,
    same_worker: bool,
    preserve_step_tags: bool,
    id: impl FnOnce() -> String,
    path: impl FnOnce() -> String,
    opt_empty_inner_flows: bool,
) -> Option<JobPayload> {
    if opt_empty_inner_flows && modules.is_empty() && modules_node.is_none() {
        return None;
    }

    if let Some(id) = modules_node {
        return Some(JobPayload::FlowNode { id, path: path() });
    }

    add_virtual_items_if_necessary(&mut modules);

    let mut failure_module = failure_module.cloned();
    if let Some(failure_module) = failure_module.as_mut() {
        failure_module.id_append(&id());
    }

    Some(JobPayload::RawFlow {
        value: FlowValue {
            modules,
            failure_module,
            same_worker,
            preserve_step_tags,
            ..Default::default()
        },
        path: Some(path()),
        restarted_from: None,
    })
}

pub fn get_path(flow_job: &MiniPulledJob, status: &FlowStatus, module: &FlowModule) -> String {
    if status
        .preprocessor_module
        .as_ref()
        .is_some_and(|x| x.id() == module.id)
    {
        format!("{}/preprocessor", flow_job.runnable_path())
    } else {
        format!("{}/{}", flow_job.runnable_path(), module.id)
    }
}

async fn compute_next_flow_transform(
    arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
    arc_last_job_result: Arc<Box<RawValue>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    flow_job: &MiniPulledJob,
    flow: &FlowValue,
    by_id: Option<IdContext>,
    db: &DB,
    module: &FlowModule,
    status: &FlowStatus,
    status_module: &FlowStatusModule,
    previous_id: &str,
    client: &AuthedClient,
    resumes: Arc<Box<RawValue>>,
    resume: Arc<Box<RawValue>>,
    approvers: Arc<Box<RawValue>>,
    is_skipped: bool,
) -> error::Result<NextFlowTransform> {
    if module.mock.is_some() && module.mock.as_ref().unwrap().enabled {
        return Ok(NextFlowTransform::Continue(
            ContinuePayload::SingleJob(JobPayloadWithTag {
                payload: JobPayload::Identity,
                tag: None,
                delete_after_use: false,
                delete_after_secs: None,
                timeout: None,
                on_behalf_of: None,
            }),
            NextStatus::NextStep,
        ));
    }
    let trivial_next_job = |payload| {
        Ok(NextFlowTransform::Continue(
            ContinuePayload::SingleJob(JobPayloadWithTag {
                payload,
                tag: None,
                delete_after_use: false,
                delete_after_secs: None,
                timeout: None,
                on_behalf_of: None,
            }),
            NextStatus::NextStep,
        ))
    };
    let delete_after_use = module.delete_after_use.unwrap_or(false);
    let delete_after_secs = module.delete_after_secs;

    tracing::debug!(id = %flow_job.id, "computing next flow transform for {:?}", &module.value);
    if is_skipped {
        return trivial_next_job(JobPayload::Identity);
    }

    match module.get_value()? {
        FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
        FlowModuleValue::Flow { path, .. } => {
            let mut payload = flow_to_payload(
                path,
                delete_after_use,
                delete_after_secs,
                &flow_job.workspace_id,
                db,
            )
            .await?;
            if let Some(nested) = nested_restart_payload(status, &module.id, None) {
                payload.payload = nested;
            }
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
                NextStatus::NextStep,
            ))
        }
        FlowModuleValue::AIAgent { tag, .. } => {
            let path = get_path(flow_job, status, module);
            let payload = JobPayload::AIAgent { path };
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(JobPayloadWithTag {
                    payload,
                    tag: tag.filter(|t| !t.trim().is_empty()),
                    delete_after_use,
                    delete_after_secs,
                    timeout: None,
                    on_behalf_of: None,
                }),
                NextStatus::NextStep,
            ))
        }
        FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
            let payload = script_to_payload(
                script_hash,
                script_path,
                db,
                flow_job,
                module,
                tag_override,
                module.apply_preprocessor,
            )
            .await?;
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
                NextStatus::NextStep,
            ))
        }
        FlowModuleValue::RawScript {
            path,
            content,
            language,
            lock,
            tag,
            concurrency_settings,
            ..
        } => {
            let path = path.unwrap_or_else(|| get_path(flow_job, status, module));

            let payload = raw_script_to_payload(
                path,
                content,
                language,
                lock,
                concurrency_settings,
                module,
                tag,
                delete_after_use,
                delete_after_secs,
            );
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
                NextStatus::NextStep,
            ))
        }
        FlowModuleValue::FlowScript {
            id, // flow_node(id).
            tag,
            language,
            concurrency_settings,
            ..
        } => {
            let path = get_path(flow_job, status, module);

            let payload = JobPayloadWithTag {
                payload: JobPayload::FlowScript {
                    id,
                    language,
                    concurrency_settings: concurrency_settings.into(),
                    cache_ttl: module.cache_ttl.map(|x| x as i32),
                    cache_ignore_s3_path: module.cache_ignore_s3_path.clone(),
                    dedicated_worker: None,
                    path,
                },
                tag: tag.clone(),
                delete_after_use,
                delete_after_secs,
                timeout: None,
                on_behalf_of: None,
            };
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
                NextStatus::NextStep,
            ))
        }
        FlowModuleValue::WhileloopFlow { modules, modules_node, squash, .. } => {
            // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg
            let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref());
            let (flow_jobs, flow_jobs_success, flow_jobs_duration) = match status_module {
                FlowStatusModule::InProgress {
                    flow_jobs: Some(flow_jobs),
                    flow_jobs_success,
                    flow_jobs_duration,
                    ..
                } => (
                    flow_jobs.clone(),
                    flow_jobs_success.clone(),
                    flow_jobs_duration.clone(),
                ),
                _ => (vec![], Some(vec![]), Some(FlowJobsDuration::new(0))),
            };
            let next_loop_idx = flow_jobs.len();

            let start_runners = next_loop_idx == 0 && squash.unwrap_or(false);
            next_loop_iteration(
                flow,
                status,
                ForloopNextIteration {
                    index: next_loop_idx,
                    itered: vec![],
                    itered_len: 0,
                    flow_jobs: flow_jobs,
                    flow_jobs_success: flow_jobs_success,
                    flow_jobs_duration: flow_jobs_duration,
                    new_args: Iter {
                        index: next_loop_idx as i32,
                        value: windmill_common::worker::to_raw_value(&next_loop_idx),
                    },
                    while_loop: true,
                },
                modules,
                modules_node,
                flow_job,
                is_simple,
                db,
                module,
                delete_after_use,
                start_runners,
            )
            .await
        }
        /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */
        FlowModuleValue::ForloopFlow {
            modules, modules_node, iterator, parallel, squash, ..
        } => {
            // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg
            let is_simple = !matches!(flow_job.kind, JobKind::FlowPreview)
                && !parallel
                && is_simple_modules(&modules, flow.failure_module.as_ref());

            // if is_simple {
            //     match value {
            //         FlowModuleValue::Script { input_transforms, .. }
            //         | FlowModuleValue::RawScript { input_transforms, .. }
            //         | FlowModuleValue::FlowScript { input_transforms, .. }
            //         | FlowModuleValue::Flow { input_transforms, .. } => {
            //             Some(input_transforms.clone())
            //         }
            //         _ => None,
            //     }
            let next_loop_status = next_forloop_status(
                status_module,
                by_id,
                flow_job,
                previous_id,
                status,
                &iterator,
                arc_last_job_result,
                resumes,
                resume,
                approvers,
                arc_flow_job_args,
                flow_env,
                client,
                &parallel,
                db,
            )
            .await?;

            match next_loop_status {
                ForLoopStatus::EmptyIterator => {
                    Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None })
                }
                ForLoopStatus::NextIteration(ns) => {
                    let start_runners = ns.index == 0 && squash.unwrap_or(false);
                    next_loop_iteration(
                        flow,
                        status,
                        ns,
                        modules,
                        modules_node,
                        flow_job,
                        is_simple,
                        db,
                        module,
                        delete_after_use,
                        start_runners,
                    )
                    .await
                }
                ForLoopStatus::ParallelIteration { itered, itered_len } => {
                    // let inner_path = Some(format!("{}/loop-parallel", flow_job.script_path(),));
                    // let value = &modules[0].get_value()?;

                    // we removed the is_simple_case
                    // if is_simple {
                    //     let payload =
                    //         payload_from_simple_module(value, db, flow_job, module, inner_path)
                    //             .await?;
                    //     ContinuePayload::ForloopJobs { n: itered.len(), payload: payload }
                    // } else {

                    let payloads = (0..itered_len)
                        .into_iter()
                        .filter_map(|i| {
                            let Some(payload) = payload_from_modules(
                                modules.clone(),
                                modules_node,
                                flow.failure_module.as_ref(),
                                flow.same_worker,
                                flow.preserve_step_tags,
                                || format!("{}-{i}", status.step),
                                || format!("{}/forloop-{i}", flow_job.runnable_path()),
                                true,
                            ) else {
                                return None;
                            };
                            Some(JobPayloadWithTag {
                                payload,
                                tag: None,
                                delete_after_use,
                                delete_after_secs,
                                timeout: None,
                                on_behalf_of: None,
                            })
                        })
                        .collect::<Vec<_>>();
                    if payloads.is_empty() {
                        return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None });
                    }
                    Ok(NextFlowTransform::Continue(
                        ContinuePayload::ParallelJobs(payloads),
                        NextStatus::AllFlowJobs {
                            branchall: None,
                            iterator: Some(FlowIterator {
                                index: 0,
                                itered_len: Some(itered_len),
                                itered: Some(itered),
                            }),
                            // we removed the is_simple_case for simple_input_transforms
                            // if is_simple {
                            //     match value {
                            //         FlowModuleValue::Script { input_transforms, .. }
                            //         | FlowModuleValue::RawScript { input_transforms, .. }
                            //         | FlowModuleValue::FlowScript { input_transforms, .. }
                            //         | FlowModuleValue::Flow { input_transforms, .. } => {
                            //             Some(input_transforms.clone())
                            //         }
                            //         _ => None,
                            //     }
                            simple_input_transforms: None,
                        },
                    ))
                }
            }
        }
        FlowModuleValue::BranchOne { branches, default, default_node } => {
            // Branch lock for nested restart: if this BranchOne is the target of a
            // nested restart, reuse the branch that was originally chosen instead of
            // re-evaluating predicates. The lock is opt-in via `restarted_from.nested`
            // being set, so plain top-level "restart at this BranchOne" still re-evaluates.
            let locked_branch = status.restarted_from.as_ref().and_then(|rf| {
                if rf.step_id == module.id && rf.nested.is_some() {
                    rf.branch_chosen.clone()
                } else {
                    None
                }
            });
            let branch = if let Some(branch) = locked_branch {
                branch
            } else {
                match status_module {
                    FlowStatusModule::WaitingForPriorSteps { .. }
                    | FlowStatusModule::WaitingForEvents { .. }
                    | FlowStatusModule::WaitingForExecutor { .. } => {
                        let mut branch_chosen = BranchChosen::Default;
                        let idcontext = get_transform_context(&flow_job, previous_id, &status);
                        let mut predicate_err: Option<Error> = None;
                        for (i, b) in branches.iter().enumerate() {
                            let pred_res = compute_bool_from_expr(
                                &b.expr,
                                arc_flow_job_args.clone(),
                                flow_env,
                                arc_last_job_result.clone(),
                                None,
                                Some(&idcontext),
                                Some(client),
                                Some((resumes.clone(), resume.clone(), approvers.clone())),
                                None,
                            )
                            .await;
                            if let Err(e) = &pred_res {
                                append_predicate_error_to_root_logs(
                                    db,
                                    root_flow_id_for(flow_job),
                                    &flow_job.workspace_id,
                                    Some(&module.id),
                                    &format!("branchone branch {i}"),
                                    &b.expr,
                                    e,
                                )
                                .await;
                            }
                            let pred = match pred_res {
                                Ok(p) => p,
                                Err(e) => {
                                    predicate_err = Some(e);
                                    break;
                                }
                            };

                            if pred {
                                branch_chosen = BranchChosen::Branch { branch: i };
                                break;
                            }
                        }
                        if let Some(e) = predicate_err {
                            return Ok(NextFlowTransform::StepFailure {
                                error: error_to_value(&e),
                            });
                        }
                        branch_chosen
                    }
                    _ => Err(Error::BadRequest(format!(
                        "Unrecognized module status for BranchOne {status_module:?}"
                    )))?,
                }
            };

            let (modules, modules_node, branch_idx) = match branch {
                BranchChosen::Default => (default, default_node, 0),
                BranchChosen::Branch { branch } => branches
                    .into_iter()
                    .nth(branch)
                    .map(|Branch { modules, modules_node, .. }| (modules, modules_node, branch + 1))
                    .ok_or_else(|| {
                        Error::BadRequest(format!(
                            "Unrecognized branch for BranchOne {status_module:?}"
                        ))
                    })?,
            };

            let Some(mut payload) = payload_from_modules(
                modules,
                modules_node,
                flow.failure_module.as_ref(),
                flow.same_worker,
                flow.preserve_step_tags,
                || status.step.to_string(),
                || format!("{}/branchone-{}", flow_job.runnable_path(), branch_idx),
                true,
            ) else {
                return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: Some(branch) });
            };
            if let Some(nested) = nested_restart_payload(status, &module.id, None) {
                payload = nested;
            }

            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(JobPayloadWithTag {
                    payload,
                    tag: None,
                    delete_after_use,
                    delete_after_secs,
                    timeout: None,
                    on_behalf_of: None,
                }),
                NextStatus::BranchChosen(branch),
            ))
        }
        FlowModuleValue::BranchAll { branches, parallel, .. } => {
            let (branch_status, flow_jobs, flow_jobs_success, flow_jobs_duration) =
                match status_module {
                    FlowStatusModule::WaitingForPriorSteps { .. }
                    | FlowStatusModule::WaitingForEvents { .. }
                    | FlowStatusModule::WaitingForExecutor { .. } => {
                        if branches.is_empty() {
                            return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None });
                        } else if parallel {
                            let len = branches.len();
                            let payloads: Vec<JobPayloadWithTag> = branches
                                .into_iter()
                                .enumerate()
                                .filter_map(|(i, Branch { modules, modules_node, .. })| {
                                    let Some(payload) = payload_from_modules(
                                        modules,
                                        modules_node,
                                        flow.failure_module.as_ref(),
                                        flow.same_worker,
                                        flow.preserve_step_tags,
                                        || format!("{}-{i}", status.step),
                                        || format!("{}/branchall-{}", flow_job.runnable_path(), i),
                                        false,
                                    ) else {
                                        return None;
                                    };
                                    Some(JobPayloadWithTag {
                                        payload,
                                        tag: None,
                                        delete_after_use,
                                        delete_after_secs,
                                        timeout: None,
                                        on_behalf_of: None,
                                    })
                                })
                                .collect::<Vec<_>>();
                            if payloads.is_empty() {
                                return Ok(NextFlowTransform::EmptyInnerFlows {
                                    branch_chosen: None,
                                });
                            }
                            return Ok(NextFlowTransform::Continue(
                                ContinuePayload::ParallelJobs(payloads),
                                NextStatus::AllFlowJobs {
                                    branchall: Some(BranchAllStatus { branch: 0, len }),
                                    iterator: None,
                                    simple_input_transforms: None,
                                },
                            ));
                        } else {
                            (
                                BranchAllStatus { branch: 0, len: branches.len() },
                                vec![],
                                Some(vec![]),
                                Some(FlowJobsDuration::new(0)),
                            )
                        }
                    }
                    FlowStatusModule::InProgress {
                        branchall: Some(BranchAllStatus { branch, len }),
                        flow_jobs: Some(flow_jobs),
                        flow_jobs_success,
                        flow_jobs_duration,
                        ..
                    } if !parallel => (
                        BranchAllStatus { branch: branch + 1, len: len.clone() },
                        flow_jobs.clone(),
                        flow_jobs_success.clone(),
                        flow_jobs_duration.clone(),
                    ),

                    _ => Err(Error::BadRequest(format!(
                        "Unrecognized module status for BranchAll {status_module:?}"
                    )))?,
                };

            let Branch { modules, modules_node, .. } = branches
                .into_iter()
                .nth(branch_status.branch)
                .ok_or_else(|| {
                    Error::BadRequest(format!(
                        "Unrecognized branch for BranchAll {status_module:?}"
                    ))
                })?;

            let Some(payload) = payload_from_modules(
                modules,
                modules_node,
                flow.failure_module.as_ref(),
                flow.same_worker,
                flow.preserve_step_tags,
                || format!("{}-{}", status.step, branch_status.branch),
                || {
                    format!(
                        "{}/branchall-{}",
                        flow_job.runnable_path(),
                        branch_status.branch
                    )
                },
                false,
            ) else {
                return Ok(NextFlowTransform::EmptyInnerFlows {
                    branch_chosen: Some(BranchChosen::Default),
                });
            };
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(JobPayloadWithTag {
                    payload,
                    tag: None,
                    delete_after_use,
                    delete_after_secs,
                    timeout: None,
                    on_behalf_of: None,
                }),
                NextStatus::NextBranchStep(NextBranch {
                    status: branch_status,
                    flow_jobs,
                    flow_jobs_success,
                    flow_jobs_duration,
                }),
            ))
        }
    }
}

async fn next_loop_iteration(
    flow: &FlowValue,
    status: &FlowStatus,
    ns: ForloopNextIteration,
    modules: Vec<FlowModule>,
    modules_node: Option<FlowNodeId>,
    flow_job: &MiniPulledJob,
    is_simple: bool,
    db: &sqlx::Pool<sqlx::Postgres>,
    module: &FlowModule,
    delete_after_use: bool,
    start_runners: bool,
) -> Result<NextFlowTransform, Error> {
    let inner_path = || format!("{}/loop-{}", flow_job.runnable_path(), ns.index);
    if is_simple {
        // Note on nested-restart and the is_simple fast path: the swap that the
        // non-simple path applies (see below) is intentionally NOT mirrored here.
        // `is_simple_modules` requires the body to be a single `script` /
        // `rawscript` / `flowscript` module (see `FlowModule::is_simple`); none of
        // those produce a flow-kind child job. Any nested-restart chain targeting
        // a leaf inside such an iteration is rejected by `resolve_nested_restart`
        // at the API layer (the leaf's container isn't a flow). Even if a chain
        // reached here via direct `JobPayload::RawFlow.restarted_from` use, the
        // resulting `JobPayload::RestartedFlow` would fail to resolve at push time
        // (script kind isn't a flow kind). So no swap is meaningful here.
        let mut value = modules[0].get_value()?;
        let simple_input_transforms = match &mut value {
            FlowModuleValue::Script { input_transforms, .. }
            | FlowModuleValue::RawScript { input_transforms, .. }
            | FlowModuleValue::FlowScript { input_transforms, .. }
            | FlowModuleValue::Flow { input_transforms, .. } => {
                Some(std::mem::take(input_transforms))
            }
            _ => None,
        };
        return Ok(NextFlowTransform::Continue(
            ContinuePayload::SingleJob(
                payload_from_simple_module(value, db, flow_job, module, inner_path()).await?,
            ),
            NextStatus::NextLoopIteration { next: ns, simple_input_transforms, start_runners },
        ));
    }

    let Some(mut payload) = payload_from_modules(
        modules,
        modules_node,
        flow.failure_module.as_ref(),
        flow.same_worker,
        flow.preserve_step_tags,
        || format!("{}-{}", status.step, ns.index),
        inner_path,
        true,
    ) else {
        return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None });
    };
    if let Some(nested) = nested_restart_payload(status, &module.id, Some(ns.index)) {
        payload = nested;
    }

    Ok(NextFlowTransform::Continue(
        ContinuePayload::SingleJob(JobPayloadWithTag {
            payload,
            tag: None,
            delete_after_use,
            delete_after_secs: None,
            timeout: None,
            on_behalf_of: None,
        }),
        NextStatus::NextLoopIteration { next: ns, simple_input_transforms: None, start_runners },
    ))
}

pub(super) fn is_simple_modules(
    modules: &Vec<FlowModule>,
    failure_module: Option<&Box<FlowModule>>,
) -> bool {
    let is_simple = modules.len() == 1
        && modules[0].is_simple()
        && modules[0].sleep.is_none()
        && modules[0].suspend.is_none()
        && modules[0].cache_ttl.is_none()
        && modules[0].retry.is_none()
        && modules[0].stop_after_if.is_none()
        && modules[0].stop_after_all_iters_if.is_none()
        && modules[0].skip_if.is_none()
        && (modules[0].mock.is_none() || modules[0].mock.as_ref().is_some_and(|m| !m.enabled))
        && failure_module.is_none();
    is_simple
}

async fn next_forloop_status(
    status_module: &FlowStatusModule,
    by_id: Option<IdContext>,
    flow_job: &MiniPulledJob,
    previous_id: &str,
    status: &FlowStatus,
    iterator: &InputTransform,
    arc_last_job_result: Arc<Box<RawValue>>,
    resumes: Arc<Box<RawValue>>,
    resume: Arc<Box<RawValue>>,
    approvers: Arc<Box<RawValue>>,
    arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
    flow_env: Option<&HashMap<String, Box<RawValue>>>,
    client: &AuthedClient,
    parallel: &bool,
    db: &DB,
) -> Result<ForLoopStatus, Error> {
    let next_loop_status = match status_module {
        FlowStatusModule::WaitingForPriorSteps { .. }
        | FlowStatusModule::WaitingForEvents { .. }
        | FlowStatusModule::WaitingForExecutor { .. } => {
            let by_id = if let Some(x) = by_id {
                x
            } else {
                get_transform_context(&flow_job, previous_id, &status)
            };
            /* Iterator is an InputTransform, evaluate it into an array. */
            let itered_raw = match iterator {
                InputTransform::Static { value } => to_raw_value(value),
                InputTransform::Ai => {
                    return Err(Error::ExecutionErr(format!(
                        "AI input transform not supported for iterator"
                    )))?
                }
                InputTransform::Javascript { expr } => {
                    let mut context = HashMap::with_capacity(5);
                    context.insert("result".to_string(), arc_last_job_result.clone());
                    context.insert("previous_result".to_string(), arc_last_job_result);
                    context.insert("resumes".to_string(), resumes);
                    context.insert("resume".to_string(), resume);
                    context.insert("approvers".to_string(), approvers);

                    eval_timeout(
                        expr.to_string(),
                        context,
                        Some(arc_flow_job_args),
                        flow_env,
                        Some(client),
                        Some(&by_id),
                        None,
                    )
                    .await?
                }
            };
            let itered = serde_json::from_str::<Vec<Box<RawValue>>>(itered_raw.get()).map_err(
                |not_array| {
                    Error::ExecutionErr(format!(
                        "Expected an array value in the iterator expression, found: {not_array}"
                    ))
                },
            )?;

            if itered.is_empty() {
                ForLoopStatus::EmptyIterator
            } else if *parallel {
                ForLoopStatus::ParallelIteration { itered_len: itered.len(), itered }
            } else if let Some(first) = itered.first() {
                let iter = Iter { index: 0 as i32, value: first.to_owned() };
                ForLoopStatus::NextIteration(ForloopNextIteration {
                    index: 0,
                    itered_len: itered.len(),
                    itered,
                    flow_jobs: vec![],
                    flow_jobs_success: Some(vec![]),
                    flow_jobs_duration: Some(FlowJobsDuration::new(0)),
                    new_args: iter,
                    while_loop: false,
                })
            } else {
                panic!("itered cannot be empty")
            }
        }

        FlowStatusModule::InProgress {
            iterator: Some(FlowIterator { itered, index, .. }),
            flow_jobs: Some(flow_jobs),
            flow_jobs_success,
            flow_jobs_duration,
            ..
        } if !*parallel => {
            // Read itered from separate table or fallback to JSONB
            let itered = read_itered_from_db(db, flow_job.id, itered).await?;

            let itered_new = if itered.is_empty() {
                // it's possible we need to re-compute the iterator Input Transforms here, in particular if the flow is being restarted inside the loop
                let by_id = if let Some(x) = by_id {
                    x
                } else {
                    get_transform_context(&flow_job, previous_id, &status)
                };
                let itered_raw = match iterator {
                    InputTransform::Static { value } => to_raw_value(value),
                    InputTransform::Javascript { expr } => {
                        let mut context = HashMap::with_capacity(5);
                        context.insert("result".to_string(), arc_last_job_result.clone());
                        context.insert("previous_result".to_string(), arc_last_job_result);
                        context.insert("resumes".to_string(), resumes);
                        context.insert("resume".to_string(), resume);
                        context.insert("approvers".to_string(), approvers);

                        eval_timeout(
                            expr.to_string(),
                            context,
                            Some(arc_flow_job_args),
                            flow_env,
                            Some(client),
                            Some(&by_id),
                            None,
                        )
                        .await?
                    }
                    InputTransform::Ai => {
                        return Err(Error::ExecutionErr(format!(
                            "AI input transform not supported for iterator"
                        )))?
                    }
                };
                serde_json::from_str::<Vec<Box<RawValue>>>(itered_raw.get()).map_err(
                    |not_array| {
                        Error::ExecutionErr(format!("Expected an array value, found: {not_array}"))
                    },
                )?
            } else {
                itered.clone()
            };
            let (index, next) = index
                 .checked_add(1)
                 .and_then(|i| itered_new.get(i).map(|next| (i, next)))
                 .with_context(|| {
                     format!("Could not find iteration number {index} restarting inside the for-loop flow. It's possible the itered-array has changed and this value isn't available anymore.")
                 })?;

            ForLoopStatus::NextIteration(ForloopNextIteration {
                index,
                itered: itered_new.clone(),
                itered_len: itered_new.len(),
                flow_jobs: flow_jobs.clone(),
                flow_jobs_success: flow_jobs_success.clone(),
                flow_jobs_duration: flow_jobs_duration.clone(),
                new_args: Iter { index: index as i32, value: next.to_owned() },
                while_loop: false,
            })
        }

        _ => Err(Error::BadRequest(format!(
            "Unrecognized module status for ForloopFlow {status_module:?}"
        )))?,
    };
    Ok(next_loop_status)
}

async fn payload_from_simple_module(
    value: FlowModuleValue,
    db: &sqlx::Pool<sqlx::Postgres>,
    flow_job: &MiniPulledJob,
    module: &FlowModule,
    inner_path: String,
) -> Result<JobPayloadWithTag, Error> {
    let delete_after_use = module.delete_after_use.unwrap_or(false);
    let delete_after_secs = module.delete_after_secs;
    Ok(match value {
        FlowModuleValue::Flow { path, .. } => {
            flow_to_payload(
                path,
                delete_after_use,
                delete_after_secs,
                &flow_job.workspace_id,
                db,
            )
            .await?
        }
        FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
            script_to_payload(
                script_hash,
                script_path,
                db,
                flow_job,
                module,
                tag_override,
                module.apply_preprocessor,
            )
            .await?
        }
        FlowModuleValue::RawScript {
            path,
            content,
            language,
            lock,
            tag,
            concurrency_settings,
            ..
        } => raw_script_to_payload(
            path.unwrap_or_else(|| inner_path),
            content,
            language,
            lock,
            concurrency_settings,
            module,
            tag,
            delete_after_use,
            delete_after_secs,
        ),
        FlowModuleValue::FlowScript {
            id, // flow_node(id).
            tag,
            language,
            concurrency_settings,
            ..
        } => JobPayloadWithTag {
            payload: JobPayload::FlowScript {
                id,
                language,
                cache_ttl: module.cache_ttl.map(|x| x as i32),
                cache_ignore_s3_path: module.cache_ignore_s3_path,
                dedicated_worker: None,
                path: inner_path,
                concurrency_settings: concurrency_settings.into(),
            },
            tag,
            delete_after_use,
            delete_after_secs,
            timeout: None, // timeout evaluation handled at higher level
            on_behalf_of: None,
        },
        _ => unreachable!("is simple flow"),
    })
}

pub fn raw_script_to_payload(
    path: String,
    content: String,
    language: windmill_common::scripts::ScriptLang,
    lock: Option<String>,
    concurrency_settings: ConcurrencySettingsWithCustom,
    module: &FlowModule,
    tag: Option<String>,
    delete_after_use: bool,
    delete_after_secs: Option<i32>,
) -> JobPayloadWithTag {
    JobPayloadWithTag {
        payload: JobPayload::Code(RawCode {
            hash: None,
            path: Some(path),
            content,
            language,
            lock,
            cache_ttl: module.cache_ttl.map(|x| x as i32),
            cache_ignore_s3_path: module.cache_ignore_s3_path,
            dedicated_worker: None,
            concurrency_settings,
            // TODO: Should this have debouncing?
            debouncing_settings: DebouncingSettings::default(),
            modules: None,
            tag: None,
        }),
        tag,
        delete_after_use,
        delete_after_secs,
        timeout: None, // timeout evaluation handled at higher level
        on_behalf_of: None,
    }
}

async fn flow_to_payload(
    path: String,
    delete_after_use: bool,
    delete_after_secs: Option<i32>,
    w_id: &str,
    db: &DB,
) -> Result<JobPayloadWithTag, Error> {
    let FlowVersionInfo { version, on_behalf_of_email, edited_by, tag, .. } =
        get_latest_flow_version_info_for_path(None, &db, w_id, &path, true).await?;
    let on_behalf_of = if let Some(email) = on_behalf_of_email {
        Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&edited_by) })
    } else {
        None
    };
    let payload = JobPayload::Flow {
        path,
        dedicated_worker: None,
        apply_preprocessor: false,
        version,
        labels: None,
    };
    Ok(JobPayloadWithTag {
        payload,
        tag,
        delete_after_use,
        delete_after_secs,
        timeout: None,
        on_behalf_of,
    })
}

pub async fn script_to_payload(
    script_hash: Option<windmill_common::scripts::ScriptHash>,
    script_path: String,
    db: &sqlx::Pool<sqlx::Postgres>,
    flow_job: &MiniPulledJob,
    module: &FlowModule,
    tag_override: Option<String>,
    apply_preprocessor: Option<bool>,
) -> Result<JobPayloadWithTag, Error> {
    let tag_override = if tag_override.as_ref().is_some_and(|x| x.trim().is_empty()) {
        None
    } else {
        tag_override
    };
    let (payload, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
        if script_hash.is_none() {
            let (jp, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
                script_path_to_payload(
                    &script_path,
                    None,
                    db.clone(),
                    &flow_job.workspace_id,
                    Some(true),
                )
                .await?;
            (
                jp,
                tag_override.to_owned().or(tag),
                delete_after_use,
                delete_after_secs,
                script_timeout,
                on_behalf_of,
            )
        } else {
            let hash = script_hash.unwrap();

            let ScriptHashInfo {
                tag,
                cache_ttl,
                language,
                dedicated_worker,
                priority,
                delete_after_use,
                delete_after_secs,
                timeout,
                on_behalf_of_email,
                created_by,
                runnable_settings:
                    ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
                ..
            } = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0)
                .await?
                .prefetch_cached(&db)
                .await?;

            let on_behalf_of = if let Some(email) = on_behalf_of_email {
                Some(OnBehalfOf {
                    email,
                    permissioned_as: username_to_permissioned_as(&created_by),
                })
            } else {
                None
            };
            (
                JobPayload::ScriptHash {
                    hash,
                    path: script_path,
                    concurrency_settings,
                    debouncing_settings,
                    cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
                    cache_ignore_s3_path: module.cache_ignore_s3_path,
                    language,
                    dedicated_worker,
                    priority,
                    apply_preprocessor: apply_preprocessor.unwrap_or(false),
                    labels: None,
                },
                tag_override.to_owned().or(tag),
                delete_after_use,
                delete_after_secs,
                timeout,
                on_behalf_of,
            )
        };
    // Module-level delete_after_secs takes precedence over script-level, then fall back to booleans
    let final_delete_after_use =
        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
    };
    Ok(JobPayloadWithTag {
        payload,
        tag,
        delete_after_use: final_delete_after_use,
        delete_after_secs: final_delete_after_secs,
        timeout: flow_step_timeout,
        on_behalf_of,
    })
}

pub fn get_transform_context(
    flow_job: &MiniPulledJob,
    previous_id: &str,
    status: &FlowStatus,
) -> IdContext {
    let steps_results: HashMap<String, JobResult> = status
        .modules
        .iter()
        .filter_map(|x| x.job_result().map(|y| (x.id(), y)))
        .collect();

    IdContext {
        flow_job: flow_job.id,
        root_flow_job: get_root_job_id(flow_job),
        steps_results,
        previous_id: previous_id.to_string(),
    }
}

// trait IntoArray: Sized {
//     fn into_array(self) -> Result<Vec<Value>, Self>;
// }

// impl IntoArray for Value {
//     fn into_array(self) -> Result<Vec<Value>, Self> {
//         match self {
//             Value::Array(array) => Ok(array),
//             not_array => Err(not_array),
//         }
//     }
// }

fn from_now(duration: Duration) -> chrono::DateTime<chrono::Utc> {
    // "This function errors when original duration is larger than
    // the maximum value supported for this type."
    chrono::Duration::from_std(duration)
        .ok()
        .and_then(|d| chrono::Utc::now().checked_add_signed(d))
        .unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
}

/// returns previous module non-zero suspend count and job, if relevant
fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid)> {
    // for a restarted job, if the restarted step is just after a suspend, don't run the suspend
    if status.restarted_from.is_some() {
        let current_step_id = flow.modules.get(status.step as usize)?.id.clone();
        if status.restarted_from.as_ref().unwrap().step_id == current_step_id {
            return None;
        }
    }
    let prev = usize::try_from(status.step)
        .ok()
        .and_then(|s| s.checked_sub(1))?;

    let suspend = flow.modules.get(prev)?.suspend.clone();
    if suspend
        .as_ref()
        .and_then(|s| s.required_events)
        .unwrap_or(0)
        == 0
    {
        return None;
    }

    if let &FlowStatusModule::Success { job, .. } = status.modules.get(prev)? {
        Some((suspend.unwrap(), job))
    } else {
        None
    }
}

// returns the result of the previous step of a running flow (if the job was successful)
pub async fn get_previous_job_result(
    db: &sqlx::Pool<sqlx::Postgres>,
    w_id: &str,
    flow_status: &FlowStatus,
) -> error::Result<Option<Box<RawValue>>> {
    let prev = usize::try_from(flow_status.step)
        .ok()
        .and_then(|s| s.checked_sub(1))
        .with_context(|| "No step preceding the current one")?;

    match flow_status.modules.get(prev) {
        Some(FlowStatusModule::Success { flow_jobs: Some(flow_jobs), .. }) => {
            Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?))
        }
        Some(FlowStatusModule::Success { job, .. }) if *job == Uuid::nil() => {
            // Empty branch — no real job was executed, return empty object
            Ok(None)
        }
        Some(FlowStatusModule::Success { job, .. }) => Ok(Some(
            sqlx::query_scalar!(
                "SELECT result AS \"result!: Json<Box<RawValue>>\"
                 FROM v2_job_completed WHERE id = $1 AND workspace_id = $2",
                job,
                w_id
            )
            .fetch_one(db)
            .await?
            .0,
        )),
        _ => Ok(None),
    }
}

#[cfg(test)]
mod tests {
    use super::extract_chat_message_from_flow_result;
    use serde_json::{json, value::to_raw_value};

    #[test]
    fn pretty_prints_full_result_when_no_override_is_present() {
        let value = json!({
            "output": "final answer",
            "metadata": { "foo": "bar" }
        });
        let result = to_raw_value(&value).unwrap();

        let message = extract_chat_message_from_flow_result(result.as_ref()).unwrap();

        assert_eq!(message, Some(serde_json::to_string_pretty(&value).unwrap()));
    }

    #[test]
    fn uses_windmill_chat_answer_when_it_is_a_string() {
        let result = to_raw_value(&json!({
            "windmill_chat_answer": "chat-visible answer",
            "output": "ignored output",
            "metadata": { "foo": "bar" }
        }))
        .unwrap();

        let message = extract_chat_message_from_flow_result(result.as_ref()).unwrap();

        assert_eq!(message, Some("chat-visible answer".to_string()));
    }

    #[test]
    fn skips_persisting_when_windmill_chat_answer_is_null() {
        let result = to_raw_value(&json!({
            "windmill_chat_answer": null,
            "output": "should not be stored"
        }))
        .unwrap();

        let message = extract_chat_message_from_flow_result(result.as_ref()).unwrap();

        assert_eq!(message, None);
    }

    #[test]
    fn coerces_scalar_windmill_chat_answer_to_a_string() {
        let result = to_raw_value(&json!({
            "windmill_chat_answer": 42,
            "output": "ignored output",
            "metadata": { "foo": "bar" }
        }))
        .unwrap();

        let message = extract_chat_message_from_flow_result(result.as_ref()).unwrap();

        assert_eq!(message, Some("42".to_string()));
    }

    #[test]
    fn pretty_prints_structured_windmill_chat_answer_only() {
        let override_value = json!({
            "text": "chat-visible answer",
            "meta": ["a", "b"]
        });
        let result = to_raw_value(&json!({
            "windmill_chat_answer": override_value,
            "output": "ignored output",
            "metadata": { "foo": "bar" }
        }))
        .unwrap();

        let message = extract_chat_message_from_flow_result(result.as_ref()).unwrap();

        assert_eq!(
            message,
            Some(serde_json::to_string_pretty(&override_value).unwrap())
        );
    }
}
