feat: load for loop jobs timeline directly from for loop flow status (#6646)

backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json

@@ -0,0 +1,19 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status SET flow_status = \n            CASE WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n                     JSONB_SET(JSONB_SET(JSONB_SET(\n                         flow_status,\n                         ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n                         $4\n                     ),\n                     ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5),\n                     ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6)\n                 ELSE\n                    JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4)\n            END\n            WHERE id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "Text",
+        "Jsonb",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3"
+}

backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json

@@ -1,17 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status SET\n                     flow_status = JSONB_SET(\n                         flow_status,\n                         ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n                         $4\n                     )\n                 WHERE id = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Uuid",
-        "Text",
-        "Jsonb"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488"
-}

backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "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>\"\n                        FROM v2_job_status WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flow_jobs_success: Json<Vec<Option<bool>>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "flow_jobs_duration: Json<FlowJobsDuration>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Int4"
+      ]
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082"
+}

backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status SET flow_status = \n                                 CASE \n                                 WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n                                    JSONB_SET(\n                                         JSONB_SET(JSONB_SET(JSONB_SET(\n                                            flow_status, \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n                                            ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n                                     )\n                                 ELSE\n                                    JSONB_SET(JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n                                         ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n                                         ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n                                     )                                 END\n                                 WHERE id = $2\n                                 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "int4",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Uuid",
+        "Text",
+        "Jsonb",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66"
+}

backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "started_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false,
+      true
+    ]
+  },
+  "hash": "8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8"
+}

backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json

@@ -1,25 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status SET\n                                     flow_status = JSONB_SET(\n                                         JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n                                         ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n                                         ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n                                     )\n                                 WHERE id = $2\n                                 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "int4",
-        "type_info": "Int4"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int4",
-        "Uuid",
-        "Text",
-        "Jsonb"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc"
-}

backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status SET flow_status = \n                                 CASE \n                                 WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n                                    JSONB_SET(\n                                         JSONB_SET(JSONB_SET(JSONB_SET(\n                                            flow_status, \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n                                            ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n                                            ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n                                            ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n                                     )\n                                 ELSE\n                                    JSONB_SET(JSONB_SET(\n                                    flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n                                         ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n                                         ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n                                     )\n                                 END\n                                 WHERE id = $2\n                                 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "int4",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Uuid",
+        "Text",
+        "Jsonb",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f"
+}

backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json

@@ -1,25 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status SET\n                                     flow_status = JSONB_SET(\n                                         JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n                                         ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n                                         ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n                                     )\n                                 WHERE id = $2\n                                 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "int4",
-        "type_info": "Int4"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int4",
-        "Uuid",
-        "Text",
-        "Jsonb"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353"
-}

backend/windmill-api/openapi.yaml

@@ -8024,6 +8024,34 @@ paths:
             application/json:
               schema: {}
 
+  /w/{workspace}/jobs_u/queue/get_started_at_by_ids:
+    post:
+      summary: get started at by ids
+      operationId: getStartedAtByIds
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: ids
+        required: true
+        content:
+          application/json:
+            schema:
+              type: array
+              items:
+                type: string
+      responses:
+        "200":
+          description: started at by ids
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: string
+                  format: date-time
+
   /w/{workspace}/jobs_u/getupdate/{id}:
     get:
       summary: get job updates
@@ -18249,7 +18277,7 @@ components:
         allow:
           type: string # comma separated permissions : "read,write,delete,list"
       required: ["pattern", "allow"]
-    
+
     GitRepositorySettings:
       type: object
       properties:

backend/windmill-api/src/jobs.rs

@@ -312,6 +312,7 @@ pub fn workspace_unauthed_service() -> Router {
             get(get_completed_job_logs_tail),
         )
         .route("/get_args/:id", get(get_args))
+        .route("/queue/get_started_at_by_ids", post(get_started_at_by_ids))
         .route("/get_flow_debug_info/:id", get(get_flow_job_debug_info))
         .route("/completed/get/:id", get(get_completed_job))
         .route("/completed/get_result/:id", get(get_completed_job_result))
@@ -1663,6 +1664,32 @@ async fn get_args(
     }
 }
 
+async fn get_started_at_by_ids(
+    Extension(db): Extension<DB>,
+    Json(mut ids): Json<Vec<Uuid>>,
+) -> JsonResult<Vec<Option<chrono::DateTime<chrono::Utc>>>> {
+    ids.truncate(100);
+
+    let started_at = sqlx::query!(
+        "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)",
+        ids.as_slice()
+    )
+    .fetch_all(&db)
+    .await?;
+
+    let as_map = started_at
+        .iter()
+        .map(|x| (x.id, x.started_at))
+        .collect::<HashMap<_, _>>();
+
+    let mut r = Vec::new();
+    for id in ids {
+        r.push(as_map.get(&id).map(|x| x.clone()).unwrap_or_default());
+    }
+
+    Ok(Json(r))
+}
+
 #[derive(Debug, sqlx::FromRow, Serialize)]
 pub struct ListableCompletedJob {
     pub r#type: String,
@@ -6868,6 +6895,7 @@ fn start_job_update_sse_stream(
                                 update.log_offset = None;
                             }
                         }
+
                         if let Some(new_stream_offset) = update.stream_offset {
                             if new_stream_offset != stream_offset.unwrap_or(0) {
                                 stream_offset = Some(new_stream_offset);

backend/windmill-common/src/flow_status.rs

@@ -113,6 +113,50 @@ pub struct FlowCleanupModule {
     pub flow_jobs_to_clean: Vec<Uuid>,
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone)]
+pub struct FlowJobsDuration {
+    pub started_at: Vec<Option<chrono::DateTime<chrono::Utc>>>,
+    pub duration_ms: Vec<Option<i64>>,
+}
+
+impl FlowJobsDuration {
+    pub fn set(&mut self, position: Option<usize>, value: &Option<FlowJobDuration>) {
+        if let Some(position) = position {
+            if position >= self.started_at.len()
+                || position >= self.duration_ms.len()
+                || value.is_none()
+            {
+                return;
+            }
+            let value = value.clone().unwrap();
+            self.started_at[position] = Some(value.started_at);
+            self.duration_ms[position] = Some(value.duration_ms);
+        }
+    }
+
+    pub fn push(&mut self, value: &Option<FlowJobDuration>) {
+        self.started_at.push(value.as_ref().map(|x| x.started_at));
+        self.duration_ms.push(value.as_ref().map(|x| x.duration_ms));
+    }
+
+    pub fn new(n: usize) -> Self {
+        Self { started_at: vec![None; n], duration_ms: vec![None; n] }
+    }
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+pub struct FlowJobDuration {
+    pub started_at: chrono::DateTime<chrono::Utc>,
+    pub duration_ms: i64,
+}
+
+impl FlowJobsDuration {
+    pub fn truncate(&mut self, n: usize) {
+        self.started_at.truncate(n);
+        self.duration_ms.truncate(n);
+    }
+}
+
 #[derive(Deserialize)]
 struct UntaggedFlowStatusModule {
     #[serde(rename = "type")]
@@ -124,6 +168,7 @@ struct UntaggedFlowStatusModule {
     iterator: Option<Iterator>,
     flow_jobs: Option<Vec<Uuid>>,
     flow_jobs_success: Option<Vec<Option<bool>>>,
+    flow_jobs_duration: Option<FlowJobsDuration>,
     branch_chosen: Option<BranchChosen>,
     branchall: Option<BranchAllStatus>,
     parallel: Option<bool>,
@@ -169,6 +214,8 @@ pub enum FlowStatusModule {
         #[serde(skip_serializing_if = "Option::is_none")]
         flow_jobs_success: Option<Vec<Option<bool>>>,
         #[serde(skip_serializing_if = "Option::is_none")]
+        flow_jobs_duration: Option<FlowJobsDuration>,
+        #[serde(skip_serializing_if = "Option::is_none")]
         branch_chosen: Option<BranchChosen>,
         #[serde(skip_serializing_if = "Option::is_none")]
         branchall: Option<BranchAllStatus>,
@@ -189,6 +236,8 @@ pub enum FlowStatusModule {
         #[serde(skip_serializing_if = "Option::is_none")]
         flow_jobs_success: Option<Vec<Option<bool>>>,
         #[serde(skip_serializing_if = "Option::is_none")]
+        flow_jobs_duration: Option<FlowJobsDuration>,
+        #[serde(skip_serializing_if = "Option::is_none")]
         branch_chosen: Option<BranchChosen>,
         #[serde(default)]
         #[serde(skip_serializing_if = "Vec::is_empty")]
@@ -209,6 +258,8 @@ pub enum FlowStatusModule {
         #[serde(skip_serializing_if = "Option::is_none")]
         flow_jobs_success: Option<Vec<Option<bool>>>,
         #[serde(skip_serializing_if = "Option::is_none")]
+        flow_jobs_duration: Option<FlowJobsDuration>,
+        #[serde(skip_serializing_if = "Option::is_none")]
         branch_chosen: Option<BranchChosen>,
         #[serde(skip_serializing_if = "Vec::is_empty")]
         failed_retries: Vec<Uuid>,
@@ -262,6 +313,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule {
                 iterator: untagged.iterator,
                 flow_jobs: untagged.flow_jobs,
                 flow_jobs_success: untagged.flow_jobs_success,
+                flow_jobs_duration: untagged.flow_jobs_duration,
                 branch_chosen: untagged.branch_chosen,
                 branchall: untagged.branchall,
                 parallel: untagged.parallel.unwrap_or(false),
@@ -279,6 +331,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule {
                     .ok_or_else(|| serde::de::Error::missing_field("job"))?,
                 flow_jobs: untagged.flow_jobs,
                 flow_jobs_success: untagged.flow_jobs_success,
+                flow_jobs_duration: untagged.flow_jobs_duration,
                 branch_chosen: untagged.branch_chosen,
                 approvers: untagged.approvers.unwrap_or_default(),
                 failed_retries: untagged.failed_retries.unwrap_or_default(),
@@ -295,6 +348,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule {
                     .ok_or_else(|| serde::de::Error::missing_field("job"))?,
                 flow_jobs: untagged.flow_jobs,
                 flow_jobs_success: untagged.flow_jobs_success,
+                flow_jobs_duration: untagged.flow_jobs_duration,
                 branch_chosen: untagged.branch_chosen,
                 failed_retries: untagged.failed_retries.unwrap_or_default(),
                 agent_actions: untagged.agent_actions,
@@ -360,6 +414,15 @@ impl FlowStatusModule {
         }
     }
 
+    pub fn flow_jobs_duration(&self) -> Option<FlowJobsDuration> {
+        match self {
+            FlowStatusModule::InProgress { flow_jobs_duration, .. } => flow_jobs_duration.clone(),
+            FlowStatusModule::Success { flow_jobs_duration, .. } => flow_jobs_duration.clone(),
+            FlowStatusModule::Failure { flow_jobs_duration, .. } => flow_jobs_duration.clone(),
+            _ => None,
+        }
+    }
+
     pub fn job_result(&self) -> Option<JobResult> {
         self.flow_jobs()
             .map(JobResult::ListJob)

backend/windmill-queue/src/jobs.rs

@@ -758,7 +758,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
     canceled_by: Option<CanceledBy>,
     flow_is_done: bool,
     duration: Option<i64>,
-) -> Result<Uuid, Error> {
+) -> Result<(Uuid, i64), Error> {
     // tracing::error!("Start");
     // let start = tokio::time::Instant::now();
 
@@ -770,7 +770,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
     }
 
     let result_columns = result_columns.as_ref();
-    let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| {
+    let (opt_uuid, duration, _skip_downstream_error_handlers) = (|| {
         commit_completed_job(
             db,
             queued_job,
@@ -799,11 +799,11 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
 
     // if scheduling next job failed, return the job_id early to ensure the job get retried after a timeout
     if let Some(job_id) = opt_uuid {
-        return Ok(job_id);
+        return Ok((job_id, duration));
     }
 
     #[cfg(feature = "cloud")]
-    apply_completed_job_cloud_usage(db, queued_job, _duration);
+    apply_completed_job_cloud_usage(db, queued_job, duration);
 
     #[cfg(feature = "enterprise")]
     apply_completed_job_error_handlers(
@@ -820,7 +820,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
 
     // tracing::error!("4 {:?}", start.elapsed());
 
-    Ok(queued_job.id)
+    Ok((queued_job.id, duration))
 }
 
 async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
@@ -4982,12 +4982,17 @@ async fn restarted_flows_resolution(
                         if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() {
                             new_flow_jobs_success.truncate(branch_or_iteration_n);
                         }
+                        let mut new_flow_jobs_timeline = module.flow_jobs_duration();
+                        if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() {
+                            new_flow_jobs_timeline.truncate(branch_or_iteration_n);
+                        }
                         truncated_modules.push(FlowStatusModule::InProgress {
                             id: module.id(),
                             job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow
                             iterator: None,
                             flow_jobs: Some(new_flow_jobs),
                             flow_jobs_success: new_flow_jobs_success,
+                            flow_jobs_duration: new_flow_jobs_timeline,
                             branch_chosen: None,
                             branchall: Some(BranchAllStatus {
                                 branch: branch_or_iteration_n - 1, // Doing minus one here as this variable reflects the latest finished job in the iteration
@@ -5022,6 +5027,10 @@ async fn restarted_flows_resolution(
                         if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() {
                             new_flow_jobs_success.truncate(branch_or_iteration_n);
                         }
+                        let mut new_flow_jobs_timeline = module.flow_jobs_duration();
+                        if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() {
+                            new_flow_jobs_timeline.truncate(branch_or_iteration_n);
+                        }
                         truncated_modules.push(FlowStatusModule::InProgress {
                             id: module.id(),
                             job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow
@@ -5031,6 +5040,7 @@ async fn restarted_flows_resolution(
                             }),
                             flow_jobs: Some(new_flow_jobs),
                             flow_jobs_success: new_flow_jobs_success,
+                            flow_jobs_duration: new_flow_jobs_timeline,
                             branch_chosen: None,
                             branchall: None,
                             parallel,

backend/windmill-worker/src/result_processor.rs

@@ -17,13 +17,7 @@ use windmill_common::otel_oss::FutureExt;
 use uuid::Uuid;
 
 use windmill_common::{
-    add_time,
-    error::{self, Error},
-    jobs::JobKind,
-    utils::WarnAfterExt,
-    worker::{to_raw_value, Connection, WORKER_GROUP},
-    worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap},
-    KillpillSender, DB,
+    add_time, error::{self, Error}, flow_status::{FlowJobDuration}, jobs::JobKind, utils::WarnAfterExt, worker::{to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB
 };
 
 #[cfg(feature = "benchmark")]
@@ -342,6 +336,7 @@ pub fn start_background_processor(
                         &w_id,
                         success,
                         Arc::new(result),
+                        None,
                         true,
                         &same_worker_tx,
                         &worker_dir,
@@ -582,6 +577,7 @@ pub async fn process_completed_job(
         let parent_job = job.parent_job.clone();
         let job_id = job.id.clone();
         let workspace_id = job.workspace_id.clone();
+        let started_at = job.started_at.clone();
 
         if job.flow_step_id.as_deref() == Some("preprocessor") {
             // Do this before inserting to `v2_job_completed` for backwards compatibility
@@ -614,7 +610,7 @@ pub async fn process_completed_job(
 
         add_time!(bench, "pre add_completed_job");
 
-        add_completed_job(
+        let (_, duration) = add_completed_job(
             db,
             &job,
             true,
@@ -642,6 +638,7 @@ pub async fn process_completed_job(
                     &workspace_id,
                     true,
                     result,
+                    started_at.map(|x| FlowJobDuration { started_at: x, duration_ms: duration }),
                     false,
                     &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(),
                     &worker_dir,
@@ -682,6 +679,7 @@ pub async fn process_completed_job(
                     &job.workspace_id,
                     false,
                     Arc::new(serde_json::value::to_raw_value(&result).unwrap()),
+                    duration.map(|x| FlowJobDuration { started_at: job.started_at.unwrap(), duration_ms: x }),
                     false,
                     &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(),
                     &worker_dir,
@@ -783,6 +781,7 @@ pub async fn handle_job_error(
             &job.workspace_id,
             false,
             Arc::new(serde_json::value::to_raw_value(&wrapped_error).unwrap()),
+            None,
             unrecoverable,
             &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).clone(),
             worker_dir,

backend/windmill-worker/src/worker_flow.rs

@@ -37,7 +37,8 @@ use windmill_common::cache::{self, RawData};
 use windmill_common::client::AuthedClient;
 use windmill_common::db::Authed;
 use windmill_common::flow_status::{
-    ApprovalConditions, FlowStatusModuleWParent, Iterator as FlowIterator, JobResult,
+    ApprovalConditions, FlowJobDuration, FlowJobsDuration, FlowStatusModuleWParent,
+    Iterator as FlowIterator, JobResult,
 };
 use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf};
 use windmill_common::jobs::{
@@ -79,6 +80,7 @@ pub async fn update_flow_status_after_job_completion(
     w_id: &str,
     success: bool,
     result: Arc<Box<RawValue>>,
+    flow_job_duration: Option<FlowJobDuration>,
     unrecoverable: bool,
     same_worker_tx: &SameWorkerSender,
     worker_dir: &str,
@@ -95,6 +97,7 @@ pub async fn update_flow_status_after_job_completion(
         job_id_for_status: job_id_for_status.clone(),
         success,
         result,
+        flow_job_duration,
         stop_early_override,
         has_triggered_error_handler: false,
     };
@@ -108,6 +111,7 @@ pub async fn update_flow_status_after_job_completion(
             &rec.job_id_for_status,
             w_id,
             rec.success,
+            rec.flow_job_duration.clone(),
             rec.result,
             unrecoverable,
             same_worker_tx,
@@ -131,6 +135,7 @@ pub async fn update_flow_status_after_job_completion(
                     &rec.job_id_for_status,
                     w_id,
                     false,
+                    rec.flow_job_duration,
                     Arc::new(to_raw_value(&Json(&WrappedError {
                         error: json!(e.to_string()),
                     }))),
@@ -180,11 +185,13 @@ pub enum UpdateFlowStatusAfterJobCompletion {
     NonLastParallelBranch,
     PreprocessingStep,
 }
+
 pub struct RecUpdateFlowStatusAfterJobCompletion {
     flow: uuid::Uuid,
     job_id_for_status: Uuid,
     success: bool,
     result: Arc<Box<RawValue>>,
+    flow_job_duration: Option<FlowJobDuration>,
     stop_early_override: Option<bool>,
     has_triggered_error_handler: bool,
 }
@@ -267,6 +274,7 @@ pub async fn update_flow_status_after_job_completion_internal(
     job_id_for_status: &Uuid,
     w_id: &str,
     mut success: bool,
+    mut flow_job_duration: Option<FlowJobDuration>,
     result: Arc<Box<RawValue>>,
     unrecoverable: bool,
     same_worker_tx: &SameWorkerSender,
@@ -508,6 +516,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                 parallel,
                 flow_jobs: Some(jobs),
                 flow_jobs_success,
+                flow_jobs_duration,
                 ..
             } if *parallel => {
                 let (nindex, len) = match (iterator, branchall) {
@@ -517,21 +526,38 @@ pub async fn update_flow_status_after_job_completion_internal(
                         } else {
                             None
                         };
-
+                        tracing::error!(
+                            "flow_job_duration: {:?}, position: {:?} flow: {:?}",
+                            flow_job_duration,
+                            position,
+                            flow
+                        );
                         let nindex = if let Some(position) = position {
                              sqlx::query_scalar!(
-                                 "UPDATE v2_job_status SET
-                                     flow_status = JSONB_SET(
-                                         JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),
+                                 "UPDATE v2_job_status SET flow_status = 
+                                 CASE 
+                                 WHEN flow_status->'modules'->$1::TEXT->'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, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::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, 'iterator', 'index'],
                                          ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb
-                                     )
+                                     )                                 END
                                  WHERE id = $2
                                  RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
                                  old_status.step,
                                  flow,
                                  position as i32,
-                                 json!(success)
+                                 json!(success),
+                                 flow_job_duration.as_ref().map(|x| json!(x.started_at)),
+                                 flow_job_duration.as_ref().map(|x| json!(x.duration_ms))
                              )
                          } else {
                              sqlx::query_scalar!(
@@ -554,6 +580,19 @@ pub async fn update_flow_status_after_job_completion_internal(
                              ))
                          })?
                          .ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress")))?;
+
+                        // 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);
+
                         tracing::info!(
                              "parallel iteration {job_id_for_status} of flow {flow} update nindex: {nindex} len: {len}",
                              nindex = nindex,
@@ -570,18 +609,33 @@ pub async fn update_flow_status_after_job_completion_internal(
 
                         let nindex = if let Some(position) = position {
                              sqlx::query_scalar!(
-                                 "UPDATE v2_job_status SET
-                                     flow_status = JSONB_SET(
-                                         JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),
+                                 "UPDATE v2_job_status SET flow_status = 
+                                 CASE 
+                                 WHEN flow_status->'modules'->$1::TEXT->'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)
+                                 json!(success),
+                                 flow_job_duration.as_ref().map(|x| json!(x.started_at)),
+                                 flow_job_duration.as_ref().map(|x| json!(x.duration_ms))
                              )
                          } else {
                              sqlx::query_scalar!(
@@ -616,7 +670,22 @@ pub async fn update_flow_status_after_job_completion_internal(
                     .execute(&mut *tx)
                     .await?;
                 if nindex == len {
-                    let mut flow_jobs_success = flow_jobs_success.clone();
+                    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 {
@@ -625,6 +694,11 @@ pub async fn update_flow_status_after_job_completion_internal(
                             }
                         }
                     }
+                    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())
@@ -698,6 +772,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                              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![],
@@ -712,6 +787,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                              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,
@@ -797,13 +873,14 @@ pub async fn update_flow_status_after_job_completion_internal(
                 && !stop_early =>
             {
                 if let Some(jobs) = flow_jobs {
-                    set_success_in_flow_job_success(
+                    set_success_and_duration_in_flow_job_success(
                         flow_jobs_success,
                         jobs,
                         job_id_for_status,
-                        &old_status,
+                        old_status.step,
                         flow,
                         success,
+                        flow_job_duration.clone(),
                         &mut tx,
                     )
                     .await?;
@@ -821,13 +898,14 @@ pub async fn update_flow_status_after_job_completion_internal(
                 && !stop_early =>
             {
                 if let Some(jobs) = flow_jobs {
-                    set_success_in_flow_job_success(
+                    set_success_and_duration_in_flow_job_success(
                         flow_jobs_success,
                         jobs,
                         job_id_for_status,
-                        &old_status,
+                        old_status.step,
                         flow,
                         success,
+                        flow_job_duration.clone(),
                         &mut tx,
                     )
                     .await?;
@@ -872,6 +950,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                 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())
@@ -884,6 +963,13 @@ pub async fn update_flow_status_after_job_completion_internal(
                     }
                 }
 
+                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, we want to set the job as failure and trigger the error handler if it exists
                 if (success
                     || (flow_jobs.is_some() && (skip_loop_failures || skip_seq_branch_failure)))
@@ -912,6 +998,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             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(),
@@ -953,6 +1040,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             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(),
@@ -1387,8 +1475,8 @@ pub async fn update_flow_status_after_job_completion_internal(
             let success = success && (!is_failure_step || result_has_recover_true(nresult.clone()));
 
             add_time!(bench, "flow status update 1");
-            if success {
-                add_completed_job(
+            let duration = if success {
+                let (_, duration) = add_completed_job(
                     db,
                     &flow_job,
                     true,
@@ -1401,8 +1489,9 @@ pub async fn update_flow_status_after_job_completion_internal(
                     None,
                 )
                 .await?;
+                duration
             } else {
-                add_completed_job(
+                let (_, duration) = add_completed_job(
                     db,
                     &flow_job,
                     false,
@@ -1419,7 +1508,11 @@ pub async fn update_flow_status_after_job_completion_internal(
                     None,
                 )
                 .await?;
-            }
+                duration
+            };
+            flow_job_duration = flow_job
+                .started_at
+                .map(|x| FlowJobDuration { started_at: x, duration_ms: duration });
         }
         true
     } else {
@@ -1469,6 +1562,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                         flow: parent_job,
                         job_id_for_status: flow,
                         success: success && !is_failure_step,
+                        flow_job_duration: flow_job_duration.clone(),
                         result: nresult.clone(),
                         stop_early_override: if stop_early {
                             Some(skip_if_stop_early)
@@ -1490,35 +1584,46 @@ fn find_flow_job_index(flow_jobs: &Vec<Uuid>, job_id_for_status: &Uuid) -> Optio
     flow_jobs.iter().position(|x| x == job_id_for_status)
 }
 
-async fn set_success_in_flow_job_success<'c>(
+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: &FlowStatus,
+    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 = JSONB_SET(
+                "UPDATE v2_job_status SET flow_status = 
+            CASE WHEN flow_status->'modules'->$1::TEXT->'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
-                     )
-                 WHERE id = $2",
-                old_status.step as i32,
+                     ),
+                     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)
+                json!(success),
+                flow_job_duration.as_ref().map(|x| json!(x.duration_ms)),
+                flow_job_duration.as_ref().map(|x| json!(x.started_at))
             )
             .execute(&mut **tx)
             .await
             .map_err(|e| {
-                Error::internal_err(format!("error while setting flow_jobs_success: {e:#}"))
+                Error::internal_err(format!(
+                    "error while setting flow_jobs_success/timeline: {e:#}"
+                ))
             })?;
         }
     }
@@ -2753,6 +2858,11 @@ async fn push_next_flow_job(
                     } 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![],
@@ -3148,6 +3258,7 @@ async fn push_next_flow_job(
                     mut flow_jobs,
                     while_loop,
                     mut flow_jobs_success,
+                    mut flow_jobs_duration,
                     ..
                 },
             ..
@@ -3159,11 +3270,15 @@ async fn push_next_flow_job(
             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: Some(FlowIterator { index, itered }),
                 flow_jobs: Some(flow_jobs),
                 flow_jobs_success,
+                flow_jobs_duration,
                 branch_chosen: None,
                 branchall: None,
                 id: status_module.id(),
@@ -3179,6 +3294,7 @@ async fn push_next_flow_job(
             iterator,
             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(),
@@ -3192,6 +3308,7 @@ async fn push_next_flow_job(
             mut flow_jobs,
             status,
             mut flow_jobs_success,
+            mut flow_jobs_duration,
             ..
         }) => {
             let uuid = one_uuid?;
@@ -3199,11 +3316,15 @@ async fn push_next_flow_job(
             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(),
@@ -3220,6 +3341,7 @@ async fn push_next_flow_job(
             iterator: None,
             flow_jobs: None,
             flow_jobs_success: None,
+            flow_jobs_duration: None,
             branch_chosen: Some(branch),
             branchall: None,
             id: status_module.id(),
@@ -3396,6 +3518,7 @@ struct ForloopNextIteration {
     itered: Vec<Box<RawValue>>,
     flow_jobs: Vec<Uuid>,
     flow_jobs_success: Option<Vec<Option<bool>>>,
+    flow_jobs_duration: Option<FlowJobsDuration>,
     new_args: Iter,
     while_loop: bool,
 }
@@ -3411,6 +3534,7 @@ struct NextBranch {
     status: BranchAllStatus,
     flow_jobs: Vec<Uuid>,
     flow_jobs_success: Option<Vec<Option<bool>>>,
+    flow_jobs_duration: Option<FlowJobsDuration>,
 }
 
 #[derive(Debug)]
@@ -3662,13 +3786,18 @@ async fn compute_next_flow_transform(
         FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => {
             // 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) = match status_module {
+            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()),
-                _ => (vec![], Some(vec![])),
+                } => (
+                    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();
             next_loop_iteration(
@@ -3679,6 +3808,7 @@ async fn compute_next_flow_transform(
                     itered: vec![],
                     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),
@@ -3877,72 +4007,78 @@ async fn compute_next_flow_transform(
             ))
         }
         FlowModuleValue::BranchAll { branches, parallel, .. } => {
-            let (branch_status, flow_jobs, flow_jobs_success) = 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,
-                                    || format!("{}-{i}", status.step),
-                                    || format!("{}/branchall-{}", flow_job.runnable_path(), i),
-                                    false,
-                                ) else {
-                                    return None;
-                                };
-                                Some(JobPayloadWithTag {
-                                    payload,
-                                    tag: None,
-                                    delete_after_use,
-                                    timeout: None,
-                                    on_behalf_of: None,
-                                })
-                            })
-                            .collect::<Vec<_>>();
-                        if payloads.is_empty() {
+            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,
+                                        || format!("{}-{i}", status.step),
+                                        || format!("{}/branchall-{}", flow_job.runnable_path(), i),
+                                        false,
+                                    ) else {
+                                        return None;
+                                    };
+                                    Some(JobPayloadWithTag {
+                                        payload,
+                                        tag: None,
+                                        delete_after_use,
+                                        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)),
+                            )
                         }
-                        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![]),
-                        )
                     }
-                }
-                FlowStatusModule::InProgress {
-                    branchall: Some(BranchAllStatus { branch, len }),
-                    flow_jobs: Some(flow_jobs),
-                    flow_jobs_success,
-                    ..
-                } if !parallel => (
-                    BranchAllStatus { branch: branch + 1, len: len.clone() },
-                    flow_jobs.clone(),
-                    flow_jobs_success.clone(),
-                ),
+                    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:?}"
-                )))?,
-            };
+                    _ => Err(Error::BadRequest(format!(
+                        "Unrecognized module status for BranchAll {status_module:?}"
+                    )))?,
+                };
 
             let Branch { modules, modules_node, .. } = branches
                 .into_iter()
@@ -3972,7 +4108,6 @@ async fn compute_next_flow_transform(
                     branch_chosen: Some(BranchChosen::Default),
                 });
             };
-
             Ok(NextFlowTransform::Continue(
                 ContinuePayload::SingleJob(JobPayloadWithTag {
                     payload,
@@ -3985,6 +4120,7 @@ async fn compute_next_flow_transform(
                     status: branch_status,
                     flow_jobs,
                     flow_jobs_success,
+                    flow_jobs_duration,
                 }),
             ))
         }
@@ -4130,6 +4266,7 @@ async fn next_forloop_status(
                     itered,
                     flow_jobs: vec![],
                     flow_jobs_success: Some(vec![]),
+                    flow_jobs_duration: Some(FlowJobsDuration::new(0)),
                     new_args: iter,
                     while_loop: false,
                 })
@@ -4142,6 +4279,7 @@ async fn next_forloop_status(
             iterator: Some(FlowIterator { itered, index }),
             flow_jobs: Some(flow_jobs),
             flow_jobs_success,
+            flow_jobs_duration,
             ..
         } if !*parallel => {
             let itered_new = if itered.is_empty() {
@@ -4192,6 +4330,7 @@ async fn next_forloop_status(
                 itered: itered_new.clone(),
                 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,
             })

cli/src/commands/sync/sync.ts

@@ -65,7 +65,10 @@ import {
 } from "../../utils/metadata.ts";
 import { OpenFlow } from "../../../gen/types.gen.ts";
 import { pushResource } from "../resource/resource.ts";
-import { newPathAssigner, PathAssigner } from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts";
+import {
+  newPathAssigner,
+  PathAssigner,
+} from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts";
 import { extractInlineScripts as extractInlineScriptsForFlows } from "../../../windmill-utils-internal/src/inline-scripts/extractor.ts";
 
 // Merge CLI options with effective settings, preserving CLI flags as overrides
@@ -158,7 +161,9 @@ async function addCodebaseDigestIfRelevant(
       try {
         parsed = yamlParseContent(path, content);
       } catch (error) {
-        log.error(`Failed to parse YAML content for codebase digest at path: ${path}`);
+        log.error(
+          `Failed to parse YAML content for codebase digest at path: ${path}`
+        );
         throw error;
       }
       if (parsed && typeof parsed == "object") {
@@ -258,7 +263,10 @@ export interface InlineScript {
   content: string;
 }
 
-export function extractInlineScriptsForApps(rec: any, pathAssigner: PathAssigner): InlineScript[] {
+export function extractInlineScriptsForApps(
+  rec: any,
+  pathAssigner: PathAssigner
+): InlineScript[] {
   if (!rec) {
     return [];
   }
@@ -349,10 +357,12 @@ function ZipFSElement(
                 flow.value.modules,
                 {},
                 SEP,
-                defaultTs,
+                defaultTs
               );
             } catch (error) {
-              log.error(`Failed to extract inline scripts for flow at path: ${p}`);
+              log.error(
+                `Failed to extract inline scripts for flow at path: ${p}`
+              );
               throw error;
             }
             for (const s of inlineScripts) {
@@ -386,9 +396,14 @@ function ZipFSElement(
             }
             let inlineScripts;
             try {
-              inlineScripts = extractInlineScriptsForApps(app?.["value"], newPathAssigner(defaultTs));
+              inlineScripts = extractInlineScriptsForApps(
+                app?.["value"],
+                newPathAssigner(defaultTs)
+              );
             } catch (error) {
-              log.error(`Failed to extract inline scripts for app at path: ${p}`);
+              log.error(
+                `Failed to extract inline scripts for app at path: ${p}`
+              );
               throw error;
             }
             for (const s of inlineScripts) {
@@ -913,13 +928,17 @@ async function compareDynFSElement(
         try {
           parsedV = JSON.parse(v);
         } catch (error) {
-          log.error(`Failed to parse new JSON content for comparison at path: ${k}`);
+          log.error(
+            `Failed to parse new JSON content for comparison at path: ${k}`
+          );
           throw error;
         }
         try {
           parsedM2 = JSON.parse(m2[k]);
         } catch (error) {
-          log.error(`Failed to parse existing JSON content for comparison at path: ${k}`);
+          log.error(
+            `Failed to parse existing JSON content for comparison at path: ${k}`
+          );
           throw error;
         }
         if (deepEqual(parsedV, parsedM2)) {
@@ -932,11 +951,11 @@ async function compareDynFSElement(
           continue;
         }
         if (!ignoreCodebaseChanges) {
-          if (before.codebase != undefined) {
+          if (before?.codebase != undefined) {
             delete before.codebase;
             m2[k] = yamlStringify(before, yamlOptions);
           }
-          if (after.codebase != undefined) {
+          if (after?.codebase != undefined) {
             if (before.codebase != after.codebase) {
               codebaseChanges[k] = after.codebase;
             }
@@ -1241,7 +1260,6 @@ export async function pull(
   opts: GlobalOptions &
     SyncOptions & { repository?: string; promotion?: string }
 ) {
-
   const originalCliOpts = { ...opts };
   opts = await mergeConfigWithConfigFile(opts);
 
@@ -1345,7 +1363,10 @@ export async function pull(
         ...(specificItems && isSpecificItem(change.path, specificItems)
           ? {
               branch_specific: true,
-              branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+              branch_specific_path: getBranchSpecificPath(
+                change.path,
+                specificItems
+              ),
             }
           : {}),
       })),
@@ -1380,7 +1401,10 @@ export async function pull(
       // Determine if this file should be written to a branch-specific path
       let targetPath = change.path;
       if (specificItems && isSpecificItem(change.path, specificItems)) {
-        const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+        const branchSpecificPath = getBranchSpecificPath(
+          change.path,
+          specificItems
+        );
         if (branchSpecificPath) {
           targetPath = branchSpecificPath;
         }
@@ -1430,12 +1454,24 @@ export async function pull(
           }
         }
         if (exts.some((e) => change.path.endsWith(e))) {
-          log.info(`Editing script content of ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
+          log.info(
+            `Editing script content of ${targetPath}${
+              targetPath !== change.path
+                ? colors.gray(` (branch-specific override for ${change.path})`)
+                : ""
+            }`
+          );
         } else if (
           change.path.endsWith(".yaml") ||
           change.path.endsWith(".json")
         ) {
-          log.info(`Editing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
+          log.info(
+            `Editing ${getTypeStrFromPath(change.path)} ${targetPath}${
+              targetPath !== change.path
+                ? colors.gray(` (branch-specific override for ${change.path})`)
+                : ""
+            }`
+          );
         }
         await Deno.writeTextFile(target, change.after);
 
@@ -1447,10 +1483,22 @@ export async function pull(
         await ensureDir(path.dirname(target));
         if (opts.stateful) {
           await ensureDir(path.dirname(stateTarget));
-          log.info(`Adding ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
+          log.info(
+            `Adding ${getTypeStrFromPath(change.path)} ${targetPath}${
+              targetPath !== change.path
+                ? colors.gray(` (branch-specific override for ${change.path})`)
+                : ""
+            }`
+          );
         }
         await Deno.writeTextFile(target, change.content);
-        log.info(`Writing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
+        log.info(
+          `Writing ${getTypeStrFromPath(change.path)} ${targetPath}${
+            targetPath !== change.path
+              ? colors.gray(` (branch-specific override for ${change.path})`)
+              : ""
+          }`
+        );
         if (opts.stateful) {
           await Deno.copyFile(target, stateTarget);
         }
@@ -1528,7 +1576,10 @@ export async function pull(
           ...(specificItems && isSpecificItem(change.path, specificItems)
             ? {
                 branch_specific: true,
-                branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+                branch_specific_path: getBranchSpecificPath(
+                  change.path,
+                  specificItems
+                ),
               }
             : {}),
         })),
@@ -1560,7 +1611,10 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) {
 
     // Check if this will be written as a branch-specific file
     if (specificItems && isSpecificItem(change.path, specificItems)) {
-      const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+      const branchSpecificPath = getBranchSpecificPath(
+        change.path,
+        specificItems
+      );
       if (branchSpecificPath) {
         displayPath = branchSpecificPath;
         branchNote = " (branch-specific)";
@@ -1569,17 +1623,26 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) {
 
     if (change.name === "added") {
       log.info(
-        colors.green(`+ ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote))
+        colors.green(
+          `+ ${getTypeStrFromPath(change.path)} ` +
+            displayPath +
+            colors.gray(branchNote)
+        )
       );
     } else if (change.name === "deleted") {
       log.info(
-        colors.red(`- ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote))
+        colors.red(
+          `- ${getTypeStrFromPath(change.path)} ` +
+            displayPath +
+            colors.gray(branchNote)
+        )
       );
     } else if (change.name === "edited") {
       log.info(
         colors.yellow(
           `~ ${getTypeStrFromPath(change.path)} ` +
-            displayPath + colors.gray(branchNote) +
+            displayPath +
+            colors.gray(branchNote) +
             (change.codebase ? ` (codebase changed)` : "")
         )
       );
@@ -1794,7 +1857,10 @@ export async function push(
         ...(specificItems && isSpecificItem(change.path, specificItems)
           ? {
               branch_specific: true,
-              branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+              branch_specific_path: getBranchSpecificPath(
+                change.path,
+                specificItems
+              ),
             }
           : {}),
       })),
@@ -1938,7 +2004,10 @@ export async function push(
                   const currentBranch = getCurrentGitBranch();
 
                   if (currentBranch && isBranchSpecificFile(resourceFilePath)) {
-                    serverPath = fromBranchSpecificPath(resourceFilePath, currentBranch);
+                    serverPath = fromBranchSpecificPath(
+                      resourceFilePath,
+                      currentBranch
+                    );
                   }
 
                   await pushResource(
@@ -1960,7 +2029,10 @@ export async function push(
               // Check if this is a branch-specific item and get the original branch-specific path
               let originalBranchSpecificPath: string | undefined;
               if (specificItems && isSpecificItem(change.path, specificItems)) {
-                originalBranchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+                originalBranchSpecificPath = getBranchSpecificPath(
+                  change.path,
+                  specificItems
+                );
               }
 
               await pushObj(
@@ -2010,7 +2082,10 @@ export async function push(
               // For branch-specific items, we read from branch-specific files but push to base server paths
               let localFilePath = change.path;
               if (specificItems && isSpecificItem(change.path, specificItems)) {
-                const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+                const branchSpecificPath = getBranchSpecificPath(
+                  change.path,
+                  specificItems
+                );
                 if (branchSpecificPath) {
                   localFilePath = branchSpecificPath;
                 }
@@ -2024,7 +2099,7 @@ export async function push(
                 opts.plainSecrets ?? false,
                 [],
                 opts.message,
-                localFilePath  // Pass the actual local file path
+                localFilePath // Pass the actual local file path
               );
 
               if (stateTarget) {
@@ -2216,7 +2291,10 @@ export async function push(
           ...(specificItems && isSpecificItem(change.path, specificItems)
             ? {
                 branch_specific: true,
-                branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+                branch_specific_path: getBranchSpecificPath(
+                  change.path,
+                  specificItems
+                ),
               }
             : {}),
         })),

frontend/openapi-ts-error-1758271586180.log

@@ -0,0 +1,28 @@
+Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25)
+
+ 8022 |           description: job args
+ 8023 |           content:
+ 8024 |             application/json:
+ 8025 |               schema: {}\
+--------------------------------^
+ 8026 | 
+ 8027 |   /w/{workspace}/jobs/queue/get_scheduled_for_by_ids:
+ParserError: Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25)
+
+ 8022 |           description: job args
+ 8023 |           content:
+ 8024 |             application/json:
+ 8025 |               schema: {}\
+--------------------------------^
+ 8026 | 
+ 8027 |   /w/{workspace}/jobs/queue/get_scheduled_for_by_ids:
+    at Object.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parsers/yaml.js:44:23)
+    at getResult (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:116:22)
+    at runNextPlugin (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:64:32)
+    at /home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:55:9
+    at new Promise (<anonymous>)
+    at Object.run (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:54:12)
+    at parseFile (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:130:38)
+    at parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:56:30)
+    at async $RefParser.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:115:28)
+    at async $RefParser.resolve (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:145:13)

frontend/src/lib/components/FlowLogViewer.svelte

@@ -19,7 +19,7 @@
 	import { twMerge } from 'tailwind-merge'
 	import FlowJobsMenu from './flows/map/FlowJobsMenu.svelte'
 	import BarsStaggered from './icons/BarsStaggered.svelte'
-	import type { GlobalIterationBounds, GraphModuleState } from './graph/model'
+	import type { GraphModuleState } from './graph/model'
 	import type { NavigationChain } from '$lib/keyboardChain'
 	import { updateLinks } from '$lib/keyboardChain'
 	import FlowLogRow from './FlowLogRow.svelte'
@@ -62,8 +62,6 @@
 		timelineAvailableWidths: Record<string, number>
 		timelinelWidth: number
 		showTimeline?: boolean
-		globalIterationBounds?: Record<string, GlobalIterationBounds>
-		loadPreviousIterations?: (key: string, amount: number) => void
 	}
 
 	let {
@@ -92,9 +90,7 @@
 		timelineNow,
 		timelineAvailableWidths = $bindable(),
 		timelinelWidth,
-		showTimeline = true,
-		globalIterationBounds,
-		loadPreviousIterations
+		showTimeline = true
 	}: Props = $props()
 
 	function getJobLink(jobId: string | undefined): string {
@@ -769,13 +765,8 @@
 														total={timelineTotal}
 														min={timelineMin}
 														items={moduleItems ?? []}
-														hasMoreIterations={globalIterationBounds?.[module.id] &&
-															(globalIterationBounds[module.id].iteration_from ?? 0) > 0}
 														now={timelineNow}
 														{timelinelWidth}
-														loadPreviousIterations={() => {
-															loadPreviousIterations?.(module.id, 20)
-														}}
 														onSelectIteration={(id) => {
 															if (
 																module.value.type !== 'forloopflow' &&
@@ -856,8 +847,6 @@
 														bind:timelineAvailableWidths
 														{timelinelWidth}
 														{showTimeline}
-														{globalIterationBounds}
-														{loadPreviousIterations}
 													/>
 												</div>
 											{/each}

frontend/src/lib/components/FlowLogViewerWrapper.svelte

@@ -1,6 +1,6 @@
 <script lang="ts">
 	import type { Job } from '$lib/gen'
-	import type { DurationStatus, GlobalIterationBounds, GraphModuleState } from './graph'
+	import type { DurationStatus, GraphModuleState } from './graph'
 	import FlowLogViewer from './FlowLogViewer.svelte'
 	import { TimelineCompute } from '$lib/timelineCompute.svelte'
 	import { onMount, untrack } from 'svelte'
@@ -11,18 +11,16 @@
 
 	interface Props {
 		job: Partial<Job>
-		localModuleStates: Record<string, GraphModuleState>
 		localDurationStatuses?: Record<string, DurationStatus>
-		globalIterationBounds?: Record<string, GlobalIterationBounds>
 		workspaceId: string | undefined
 		render: boolean
+		localModuleStates: Record<string, GraphModuleState>
 		onSelectedIteration?: (
 			detail:
 				| { id: string; index: number; manuallySet: true; moduleId: string }
 				| { manuallySet: false; moduleId: string }
 		) => Promise<void>
 		mode?: 'flow' | 'aiagent'
-		loadPreviousIterations?: (key: string, amount: number) => void
 	}
 
 	let {
@@ -32,9 +30,7 @@
 		workspaceId,
 		render,
 		onSelectedIteration,
-		mode = 'flow',
-		globalIterationBounds,
-		loadPreviousIterations
+		mode = 'flow'
 	}: Props = $props()
 
 	// State for tracking expanded rows - using Record to allow explicit control
@@ -194,7 +190,5 @@
 		{timelineNow}
 		bind:timelineAvailableWidths
 		{timelinelWidth}
-		{globalIterationBounds}
-		{loadPreviousIterations}
 	/>
 </div>

frontend/src/lib/components/FlowStatusViewer.svelte

@@ -66,8 +66,6 @@
 	let retryStatus = $state({ val: {} })
 	let globalRefreshes: Record<string, ((clear, root) => Promise<void>)[]> = $state({})
 
-	let globalIterationBounds = $state({})
-
 	setContext<FlowStatusViewerContext>('FlowStatusViewer', {
 		flowStateStore,
 		suspendStatus,
@@ -90,7 +88,6 @@
 			retryStatus.val = {}
 			suspendStatus.val = {}
 			globalRefreshes = {}
-			globalIterationBounds = {}
 			flowStateStore.val = {}
 			localDurationStatuses = {}
 			localModuleStates = {}
@@ -130,7 +127,6 @@
 			onJobsLoaded?.({ job, force })
 		}}
 		globalModuleStates={[]}
-		{globalIterationBounds}
 		bind:localModuleStates
 		bind:selectedNode={selectedJobStep}
 		bind:localDurationStatuses

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -20,12 +20,7 @@
 	import { onDestroy } from 'svelte'
 	import { Badge, Button, Skeleton, Tab } from './common'
 	import Tabs from './common/tabs/Tabs.svelte'
-	import {
-		type DurationStatus,
-		type FlowStatusViewerContext,
-		type GlobalIterationBounds,
-		type GraphModuleState
-	} from './graph'
+	import { type DurationStatus, type FlowStatusViewerContext, type GraphModuleState } from './graph'
 	import ModuleStatus from './ModuleStatus.svelte'
 	import { clone, isScriptPreview, msToSec, readFieldsRecursively, truncateRev } from '$lib/utils'
 	import JobArgs from './JobArgs.svelte'
@@ -72,6 +67,12 @@
 					moduleId: string
 					flowJobs: string[]
 					flowJobsSuccess: (boolean | undefined)[]
+					flowJobsDuration:
+						| {
+								started_at?: (string | undefined)[]
+								duration_ms?: (number | undefined)[]
+						  }
+						| undefined
 					length: number
 					branchall?: boolean
 			  }
@@ -83,7 +84,6 @@
 		selectedNode?: string | undefined
 		globalModuleStates: Record<string, GraphModuleState>[]
 		globalDurationStatuses?: Record<string, DurationStatus>[]
-		globalIterationBounds: Record<string, GlobalIterationBounds>
 		isSelectedBranch?: boolean
 		isSubflow?: boolean
 		reducedPolling?: boolean
@@ -138,7 +138,6 @@
 		selectedNode = $bindable(undefined),
 		globalModuleStates,
 		globalDurationStatuses = [],
-		globalIterationBounds,
 		updateRecursiveRefreshFn = undefined,
 		isSelectedBranch = true,
 		isSubflow = false,
@@ -402,15 +401,30 @@
 		}
 	}
 
-	function setDurationStatusByJob(key: string, id: string, value: any) {
-		if (!deepEqual(localDurationStatuses[key]?.byJob[id], value)) {
-			localDurationStatuses[key].byJob[id] = value
+	function setDurationStatusByJob(
+		key: string,
+		id: string,
+		value: DurationStatus['byJob'][string],
+		overwrite: boolean = false
+	) {
+		if (!deepEqual(localDurationStatuses[key]?.byJob?.[id], value)) {
+			if (localDurationStatuses[key]?.byJob == undefined) {
+				localDurationStatuses[key] = { byJob: {} }
+			}
+			localDurationStatuses[key].byJob = {
+				[id]: value,
+				...(overwrite ? {} : (localDurationStatuses[key].byJob ?? {}))
+			}
+
 			globalDurationStatuses.forEach((s) => {
-				s[key].byJob[id] = value
+				s[key].byJob = { [id]: value, ...(overwrite ? {} : (s[key].byJob ?? {})) }
 			})
 			if (prefix) {
 				subflowParentsDurationStatuses.forEach((s) => {
-					s[buildSubflowKey(key, prefix)].byJob[id] = value
+					s[buildSubflowKey(key, prefix)].byJob = {
+						[id]: value,
+						...(overwrite ? {} : (s[buildSubflowKey(key, prefix)].byJob ?? {}))
+					}
 				})
 			}
 		}
@@ -461,6 +475,34 @@
 		}
 	}
 
+	function updateDurationStatuses(
+		key: string,
+		durationStatuses: Record<string, DurationStatus['byJob'][string]>
+	) {
+		if (localDurationStatuses[key] == undefined) {
+			localDurationStatuses[key] = { byJob: {} }
+		}
+		localDurationStatuses[key].byJob = durationStatuses
+		globalDurationStatuses.forEach((s) => {
+			if (s[key] == undefined) {
+				s[key] = { byJob: {} }
+			}
+			s[key].byJob = durationStatuses
+		})
+		if (prefix) {
+			subflowParentsDurationStatuses.forEach((s) => {
+				if (s[key] == undefined) {
+					s[key] = { byJob: {} }
+				}
+				s[key].byJob = durationStatuses
+			})
+		}
+	}
+
+	let jobMissingStartedAt: Record<string, number | 'P'> = {}
+	let lastSelectedLoopSwitch: number | undefined
+	let selectedLoopSwitchTimeout: number | undefined = undefined
+
 	function updateInnerModules() {
 		if (localModuleStates) {
 			innerModules?.forEach((mod, i) => {
@@ -502,7 +544,7 @@
 									mod.branch_chosen.type == 'default' ? 0 : (mod.branch_chosen.branch ?? 0) + 1
 							}
 						: {}
-					console.debug('updateInnerModules', mod.id, mod.type, branchChosen)
+					// console.debug('updateInnerModules', mod.id, mod.type, branchChosen)
 					setModuleState(
 						mod.id ?? '',
 						{
@@ -515,14 +557,114 @@
 					setModuleState(mod.id ?? '', {}, true)
 				}
 
-				if (mod.flow_jobs_success || mod.flow_jobs) {
+				if (mod.flow_jobs_success || mod.flow_jobs_duration || mod.flow_jobs) {
 					setModuleState(mod.id ?? '', {
 						flow_jobs_success: mod.flow_jobs_success,
+						flow_jobs_duration: mod.flow_jobs_duration,
 						flow_jobs: mod.flow_jobs,
 						iteration_total: mod.iterator?.itered?.length ?? mod.flow_jobs?.length
 					})
 				}
 
+				if (mod.flow_jobs_duration && mod.flow_jobs) {
+					let key = buildSubflowKey(mod.id ?? '', prefix)
+					let durationStatuses = Object.fromEntries(
+						mod.flow_jobs.map((flowJobId, idx) => {
+							let started_at_str = mod.flow_jobs_duration?.started_at?.[idx]
+							let started_at = started_at_str ? new Date(started_at_str).getTime() : undefined
+							let duration_ms = mod.flow_jobs_duration?.duration_ms?.[idx]
+							if (started_at == undefined) {
+								let missingStartedAt = jobMissingStartedAt[flowJobId]
+								if (missingStartedAt != 'P') {
+									started_at = missingStartedAt
+								}
+							} else {
+								delete jobMissingStartedAt[flowJobId]
+							}
+							return [
+								flowJobId,
+								{
+									created_at: started_at,
+									started_at: started_at,
+									duration_ms: duration_ms
+								}
+							]
+						})
+					)
+					let missingStartedAtIds = Object.keys(durationStatuses)
+						.filter(
+							(id) => durationStatuses[id].created_at == undefined && jobMissingStartedAt[id] != 'P'
+						)
+						.slice(0, 100)
+
+					updateDurationStatuses(key, durationStatuses)
+
+					if (missingStartedAtIds.length > 0) {
+						missingStartedAtIds.forEach((id) => {
+							jobMissingStartedAt[id] = 'P'
+						})
+						JobService.getStartedAtByIds({
+							workspace: workspaceId ?? $workspaceStore ?? '',
+							requestBody: missingStartedAtIds
+						})
+							.then((jobs) => {
+								let lastStarted: string | undefined = undefined
+								let anySet = false
+								let nDurationStatuses = localDurationStatuses[key]?.byJob
+								missingStartedAtIds.forEach((id, idx) => {
+									const startedAt = jobs[idx]
+									const time = startedAt ? new Date(startedAt).getTime() : undefined
+									if (time) {
+										jobMissingStartedAt[id] = time
+									} else {
+										delete jobMissingStartedAt[id]
+									}
+									if (nDurationStatuses && time) {
+										if (!nDurationStatuses[id]?.duration_ms) {
+											anySet = true
+											lastStarted = id
+											nDurationStatuses[id] = {
+												created_at: time,
+												started_at: time
+											}
+										}
+									}
+								})
+								if (anySet) {
+									updateDurationStatuses(key, nDurationStatuses)
+									selectedLoopSwitchTimeout && clearTimeout(selectedLoopSwitchTimeout)
+									function setSelectedLoopSwitch() {
+										if (lastStarted) {
+											let position = mod.flow_jobs?.indexOf(lastStarted)
+											if (position != undefined) {
+												lastSelectedLoopSwitch = new Date().getTime()
+												console.log('setSelectedLoopSwitch', position, lastStarted)
+												setIteration(position, lastStarted, false, mod.id ?? '', true)
+											}
+										}
+									}
+
+									if (
+										lastSelectedLoopSwitch &&
+										new Date().getTime() - lastSelectedLoopSwitch < 3000
+									) {
+										selectedLoopSwitchTimeout = setTimeout(() => {
+											setSelectedLoopSwitch()
+										}, 2000)
+									} else {
+										console.log('setSelectedLoopSwitch')
+										setSelectedLoopSwitch()
+									}
+								}
+							})
+							.catch((e) => {
+								console.error(`Could not load inner module duration status for job ${mod.job}`, e)
+							})
+					} else {
+						setIteration(0, mod.flow_jobs?.[0] ?? '', false, mod.id ?? '', true)
+					}
+				}
+
 				if (mod.agent_actions && mod.id) {
 					setModuleState(mod.id, {
 						agent_actions: mod.agent_actions
@@ -636,7 +778,6 @@
 			flowTimeline?.reset()
 			timeout && clearTimeout(timeout)
 			innerModules = undefined
-			console.log('updateJobId', jobId)
 			if (flowJobIds) {
 				let modId = flowJobIds?.moduleId ?? ''
 
@@ -649,11 +790,6 @@
 						x[prefixed] = { byJob: {} }
 					}
 				})
-
-				globalIterationBounds[prefixed] = {
-					iteration_from: flowJobIds?.branchall ? 0 : Math.max(flowJobIds.flowJobs.length - 20, 0),
-					iteration_total: flowJobIds?.length
-				}
 			} else {
 				recursiveRefresh = {}
 				localDurationStatuses = {}
@@ -711,10 +847,15 @@
 					},
 					force
 				)
-				setDurationStatusByJob(id, job.id, {
-					created_at: job.created_at ? new Date(job.created_at).getTime() : undefined,
-					started_at
-				})
+				setDurationStatusByJob(
+					id,
+					job.id,
+					{
+						created_at: job.created_at ? new Date(job.created_at).getTime() : undefined,
+						started_at
+					},
+					true
+				)
 			} else {
 				const parent_module = mod['parent_module']
 
@@ -735,6 +876,7 @@
 						started_at: started_at,
 						flow_jobs: mod.flow_jobs,
 						flow_jobs_success: mod.flow_jobs_success,
+						flow_jobs_duration: mod.flow_jobs_duration,
 						iteration_total: mod.iterator?.itered?.length,
 						retries: mod?.failed_retries?.length,
 						skipped: mod.skipped,
@@ -744,11 +886,16 @@
 					force
 				)
 
-				setDurationStatusByJob(id, job.id, {
-					created_at: job.created_at ? new Date(job.created_at).getTime() : undefined,
-					started_at,
-					duration_ms: job['duration_ms']
-				})
+				setDurationStatusByJob(
+					id,
+					job.id,
+					{
+						created_at: job.created_at ? new Date(job.created_at).getTime() : undefined,
+						started_at,
+						duration_ms: job['duration_ms']
+					},
+					true
+				)
 			}
 		}
 	}
@@ -858,6 +1005,7 @@
 				started_at,
 				flow_jobs: flowJobIds?.flowJobs,
 				flow_jobs_success: flowJobIds?.flowJobsSuccess,
+				flow_jobs_duration: flowJobIds?.flowJobsDuration,
 				iteration_total: flowJobIds?.length,
 				duration_ms: undefined
 			}
@@ -880,17 +1028,29 @@
 				}
 			}
 			setModuleState(modId, v, force, true)
-			if (jobLoaded.type == 'QueuedJob') {
-				setDurationStatusByJob(modId, job_id, {
-					created_at,
-					started_at
-				})
-			} else if (jobLoaded.type == 'CompletedJob') {
-				setDurationStatusByJob(modId, job_id, {
-					created_at,
-					started_at,
-					duration_ms: jobLoaded.duration_ms
-				})
+			if (innerModule?.type == 'branchall') {
+				if (jobLoaded.type == 'QueuedJob') {
+					setDurationStatusByJob(
+						modId,
+						job_id,
+						{
+							created_at,
+							started_at
+						},
+						false
+					)
+				} else if (jobLoaded.type == 'CompletedJob') {
+					setDurationStatusByJob(
+						modId,
+						job_id,
+						{
+							created_at,
+							started_at,
+							duration_ms: jobLoaded.duration_ms
+						},
+						false
+					)
+				}
 			}
 
 			if (jobLoaded.job_kind == 'script' || isScriptPreview(jobLoaded.job_kind)) {
@@ -910,15 +1070,6 @@
 
 	let flowTimeline: FlowTimeline | undefined = $state()
 
-	function loadPreviousIters(innerKey: string, lenToAdd: number) {
-		let key = buildSubflowKey(innerKey, prefix)
-		if (globalIterationBounds[key]) {
-			globalIterationBounds[key].iteration_from =
-				(globalIterationBounds[key]?.iteration_from ?? 0) - lenToAdd
-		}
-		// updateSlicedListJobIds()
-	}
-
 	let stepDetail: FlowModule | string | undefined = $state(undefined)
 
 	let storedListJobs: Record<number, Job> = $state({})
@@ -941,15 +1092,25 @@
 		}
 	}
 
+	export type FlowModuleForTimeline = {
+		id: string
+		type: FlowModuleValue['type']
+	}
+
 	function allModulesForTimeline(
 		modules: FlowModule[],
 		expandedSubflows: Record<string, FlowModule[]>
-	): string[] {
-		const ids = dfs(modules, (x) => x.id, { skipToolNodes: true })
+	): FlowModuleForTimeline[] {
+		const ids = dfs(modules, (x) => ({ id: x.id, type: x.value.type }) as FlowModuleForTimeline, {
+			skipToolNodes: true
+		})
 
-		function rec(ids: string[], prefix: string | undefined): string[] {
+		function rec(
+			ids: FlowModuleForTimeline[],
+			prefix: string | undefined
+		): FlowModuleForTimeline[] {
 			return ids.concat(
-				ids.flatMap((id) => {
+				ids.flatMap(({ id, type }) => {
 					let fms = expandedSubflows[id]
 					let oid = id.split(':').pop()
 					if (!oid) {
@@ -960,7 +1121,10 @@
 						? rec(
 								dfs(
 									fms,
-									(x) => (x.id.startsWith('subflow:') ? x.id : buildSubflowKey(x.id, nprefix)),
+									(x) => ({
+										id: x.id.startsWith('subflow:') ? x.id : buildSubflowKey(x.id, nprefix),
+										type: x.value.type
+									}),
 									{ skipToolNodes: true }
 								),
 								nprefix
@@ -1113,32 +1277,6 @@
 			<div class="h-8" />
 		{/if} -->
 		{#if isListJob}
-			{@const sliceFrom =
-				globalIterationBounds[buildSubflowKey(flowJobIds?.moduleId ?? '', prefix)]
-					?.iteration_from ?? 0}
-			{@const lenToAdd = Math.min(20, sliceFrom)}
-			{#if (flowJobIds?.flowJobs.length ?? 0) > 20 && lenToAdd > 0}
-				{@const allToAdd = (flowJobIds?.length ?? 0) - sliceFrom}
-				<p class="text-tertiary italic text-xs">
-					For performance reasons, only the last 20 items are shown by default <button
-						class="text-primary underline ml-4"
-						onclick={() => {
-							loadPreviousIters(flowJobIds?.moduleId ?? '', lenToAdd)
-						}}
-						>Load {lenToAdd} prior
-					</button>
-					{#if allToAdd > 0 && allToAdd > lenToAdd}
-						{sliceFrom}
-						<button
-							class="text-primary underline ml-4"
-							onclick={() => {
-								loadPreviousIters(flowJobIds?.moduleId ?? '', allToAdd)
-							}}
-							>Load {allToAdd} prior
-						</button>
-					{/if}
-				</p>
-			{/if}
 			{#if render}
 				<div class="w-full h-full border rounded-sm bg-surface p-1 overflow-auto">
 					<DisplayResult
@@ -1188,9 +1326,6 @@
 			style="min-height: {minTabHeight}px"
 		>
 			{#if isListJob}
-				{@const sliceFrom =
-					globalIterationBounds[buildSubflowKey(flowJobIds?.moduleId ?? '', prefix)]
-						?.iteration_from ?? 0}
 				<h3 class="text-md leading-6 font-bold text-tertiary border-b mb-4">
 					Subflows ({flowJobIds?.flowJobs.length})
 				</h3>
@@ -1233,7 +1368,7 @@
 								</span>
 							</Button>
 						{/if}
-						{#if j >= sliceFrom || forloop_selected == loopJobId}
+						{#if forloop_selected == loopJobId || innerModule?.type == 'branchall'}
 							{@const forloopIsSelected =
 								forloop_selected == loopJobId ||
 								(innerModule?.type != 'forloopflow' && innerModule?.type != 'whileloopflow')}
@@ -1274,7 +1409,6 @@
 									{onResultStreamUpdate}
 									graphTabOpen={selected == 'graph' && graphTabOpen}
 									isNodeSelected={forloop_selected == loopJobId}
-									{globalIterationBounds}
 									toolCallStore={{
 										getStoredToolCallJob: (storeKey: string) =>
 											toolCallStore?.getStoredToolCallJob(forLoopStoreKeyPrefix + storeKey),
@@ -1370,7 +1504,6 @@
 											{onResultStreamUpdate}
 											graphTabOpen={selected == 'graph' && graphTabOpen}
 											isNodeSelected={retry_selected == failedRetry}
-											{globalIterationBounds}
 										/>
 									</div>
 								{/each}
@@ -1385,7 +1518,6 @@
 										globalModuleStates={[]}
 										{updateGlobalRefresh}
 										globalDurationStatuses={[]}
-										{globalIterationBounds}
 										prefix={buildPrefix(prefix, mod.id ?? '')}
 										subflowParentsGlobalModuleStates={[
 											localModuleStates,
@@ -1433,6 +1565,7 @@
 													moduleId: mod.id ?? '',
 													flowJobs: mod.flow_jobs,
 													flowJobsSuccess: mod.flow_jobs_success ?? [],
+													flowJobsDuration: mod.flow_jobs_duration,
 													length: mod.iterator?.itered?.length ?? mod.flow_jobs.length,
 													branchall: job?.raw_flow?.modules?.[i]?.value?.type == 'branchall'
 												}
@@ -1448,7 +1581,6 @@
 										{onResultStreamUpdate}
 										graphTabOpen={selected == 'graph' && graphTabOpen}
 										isNodeSelected={localModuleStates?.[selectedNode ?? '']?.job_id == mod.job}
-										{globalIterationBounds}
 										{toolCallStore}
 									/>
 									{#if mod.agent_actions && mod.agent_actions.length > 0 && mod.id}
@@ -1518,7 +1650,6 @@
 														graphTabOpen={selected == 'graph' && graphTabOpen}
 														isNodeSelected={localModuleStates?.[toolCallId]?.job_id ==
 															agentAction.job_id}
-														{globalIterationBounds}
 													/>
 												{/if}
 											{/if}
@@ -1551,10 +1682,6 @@
 					{workspaceId}
 					{render}
 					{onSelectedIteration}
-					{globalIterationBounds}
-					loadPreviousIterations={(key, amount) => {
-						loadPreviousIters(key, amount)
-					}}
 				/>
 			</div>
 		{:else if selected == 'assets' && render}
@@ -1660,6 +1787,8 @@
 						</Tabs>
 						{#if rightColumnSelect == 'timeline'}
 							<FlowTimeline
+								{localModuleStates}
+								{onSelectedIteration}
 								selfWaitTime={job?.self_wait_time_ms}
 								aggregateWaitTime={job?.aggregate_wait_time_ms}
 								flowDone={job?.['success'] != undefined}
@@ -1669,11 +1798,7 @@
 									expandedSubflows ?? {}
 								)}
 								buildSubflowKey={(key) => buildSubflowKey(key, prefix)}
-								{globalIterationBounds}
 								durationStatuses={localDurationStatuses}
-								decreaseIterationFrom={(key, amount) => {
-									loadPreviousIters(key, amount)
-								}}
 							/>
 						{:else if rightColumnSelect == 'node_status'}
 							<div class="pt-2 grow flex flex-col">

frontend/src/lib/components/FlowTimeline.svelte

@@ -3,15 +3,25 @@
 	import { Loader2 } from 'lucide-svelte'
 	import TimelineBar from './TimelineBar.svelte'
 	import WaitTimeWarning from './common/waitTimeWarning/WaitTimeWarning.svelte'
-	import type { GlobalIterationBounds } from './graph'
 	import { TimelineCompute } from '$lib/timelineCompute.svelte'
 	import { onMount, untrack } from 'svelte'
 	import OnChange from './common/OnChange.svelte'
+	import VirtualList from '@tutorlatin/svelte-tiny-virtual-list'
+	import type { GraphModuleState } from './graph'
+	import FlowJobsMenu from './flows/map/FlowJobsMenu.svelte'
+	import type { FlowModuleForTimeline } from './FlowStatusViewerInner.svelte'
 
 	interface Props {
 		selfWaitTime?: number | undefined
 		aggregateWaitTime?: number | undefined
-		flowModules: string[]
+		flowModules: FlowModuleForTimeline[]
+		localModuleStates: Record<string, GraphModuleState>
+
+		onSelectedIteration?: (
+			detail:
+				| { id: string; index: number; manuallySet: true; moduleId: string }
+				| { manuallySet: false; moduleId: string }
+		) => Promise<void>
 		durationStatuses: Record<
 			string,
 			{
@@ -19,9 +29,7 @@
 			}
 		>
 		flowDone?: boolean
-		decreaseIterationFrom?: (key: string, amount: number) => void
 		buildSubflowKey: (key: string) => string
-		globalIterationBounds: Record<string, GlobalIterationBounds>
 	}
 
 	let {
@@ -30,21 +38,28 @@
 		flowModules,
 		durationStatuses,
 		flowDone = false,
-		decreaseIterationFrom,
-		buildSubflowKey,
-		globalIterationBounds
+		localModuleStates,
+		onSelectedIteration
 	}: Props = $props()
 
 	let timelineCompute = $state<TimelineCompute | undefined>(undefined)
+	const flowModulesIds = $derived(flowModules.map(({ id }) => id))
 
 	// Initialize timeline compute when we have duration statuses
 	onMount(() => {
-		timelineCompute = new TimelineCompute(flowModules, durationStatuses, flowDone)
+		timelineCompute = new TimelineCompute(flowModulesIds, durationStatuses, flowDone)
 		return () => {
 			timelineCompute?.destroy()
 		}
 	})
 
+	$effect(() => {
+		flowDone
+		untrack(() => {
+			timelineCompute?.setFlowDone(flowDone)
+		})
+	})
+
 	// Derived timeline values
 	const min = $derived(timelineCompute?.min ?? undefined)
 	const max = $derived(timelineCompute?.max ?? undefined)
@@ -56,21 +71,15 @@
 		timelineCompute?.reset()
 	}
 
-	$effect(() => {
-		flowDone
-		untrack(() => {
-			timelineCompute?.setFlowDone(flowDone)
-		})
-	})
+	const barHeight = 32
 </script>
 
 <OnChange
 	key={durationStatuses}
 	onChange={() => {
-		timelineCompute?.updateInputs(flowModules, durationStatuses, flowDone)
+		timelineCompute?.updateInputs(flowModulesIds, durationStatuses, flowDone)
 	}}
 />
-
 {#if items}
 	<div class="divide-y border-b">
 		<div class="px-2 py-2 grid grid-cols-12 w-full"
@@ -81,7 +90,7 @@
 					>
 				{/if}<div class="flex gap-1 items-center font-mono"
 					>{max ? displayDate(new Date(max), true) : ''}{#if !max && min}{#if now}
-							{msToSec(now - min, 3)}s
+							{msToSec(now - min, 1)}s
 						{/if}<Loader2 size={14} class="animate-spin" />{/if}</div
 				></div
 			>
@@ -109,61 +118,87 @@
 				/>
 			</div>
 		{/if}
-		{#each Object.values(flowModules) as k (k)}
-			{@const iterationFrom = globalIterationBounds[buildSubflowKey(k)]?.iteration_from ?? 0}
-			<div class="overflow-auto max-h-60 shadow-inner dark:shadow-gray-700 relative">
-				{#if iterationFrom > 0}
-					<div class="w-full flex flex-row-reverse sticky top-0">
-						<button
-							class="!text-secondary underline mr-2 text-2xs text-right whitespace-nowrap"
-							onclick={() => {
-								decreaseIterationFrom?.(k, 20)
-							}}
-							>Viewing iterations {iterationFrom} to {globalIterationBounds[buildSubflowKey(k)]
-								?.iteration_total}. Load more
-						</button>
-					</div>
-				{/if}
-
+		{#each flowModules as { id: k, type: typ } (k)}
+			{@const subItems = items?.[k]?.filter((x) => x.created_at && x.started_at)}
+			<div class="shadow-inner dark:shadow-gray-700 relative">
 				<div class="px-2 py-2 grid grid-cols-6 w-full">
-					<div class="truncate">{k.startsWith('subflow:') ? k.substring(8) : k}</div>
-					<div class="col-span-5 flex min-h-6">
+					<div class="truncate"
+						>{k.startsWith('subflow:') ? k.substring(8) : k}
+						{#if localModuleStates[k]?.selectedForloop && (typ == 'forloopflow' || typ == 'whileloopflow')}
+							<span class="text-xs font-mono font-medium inline-flex items-center -my-2">
+								<button onclick={(e) => e.stopPropagation()}>
+									<FlowJobsMenu
+										moduleId={k}
+										id={k}
+										{onSelectedIteration}
+										flowJobsSuccess={localModuleStates[k]?.flow_jobs_success}
+										flowJobs={localModuleStates[k]?.flow_jobs}
+										selected={localModuleStates[k]?.selectedForloopIndex ?? 0}
+										selectedManually={localModuleStates[k]?.selectedForLoopSetManually ?? false}
+										showIcon={false}
+									/>
+								</button>
+							</span>
+						{/if}
+					</div>
+					<div class="col-span-5 flex">
+						{#if subItems?.length > 1}
+							<div class="text-xs text-secondary absolute top-1 right-2">
+								{subItems?.length} jobs
+							</div>
+						{/if}
 						{#if min && total}
-							<div class="flex flex-col gap-2 w-full p-2 ml-4">
-								{#each items?.[k] ?? [] as b}
-									{@const waitingLen = b?.created_at
-										? b.started_at
-											? b.started_at - b?.created_at
-											: b.duration_ms
-												? 0
-												: now - b?.created_at
-										: 0}
-									<div class="flex w-full">
-										<TimelineBar
-											position="left"
-											id={b?.id}
-											{total}
-											{min}
-											gray
-											started_at={b.created_at}
-											len={waitingLen < 100 ? 0 : waitingLen - 100}
-											running={b?.started_at == undefined}
-										/>
-										{#if b.started_at}
+							<VirtualList
+								width="100%"
+								height={Math.min(400, (subItems?.length ?? 0) * barHeight)}
+								itemCount={subItems?.length ?? 0}
+								itemSize={barHeight}
+								getKey={(index) => subItems?.[index]?.id}
+							>
+								{#snippet item({ index, style })}
+									{@const b = subItems?.[index]}
+									{#if b?.created_at}
+										<!-- <div class="text-xs text-secondary">{JSON.stringify(b)}</div> -->
+										{@const waitingLen = b?.created_at
+											? b.started_at
+												? b.started_at - b?.created_at
+												: b.duration_ms
+													? 0
+													: now - b?.created_at
+											: 0}
+										<div class="flex w-full p-1 pb-2 pl-12" {style}>
 											<TimelineBar
-												position={waitingLen < 100 ? 'center' : 'right'}
+												position="left"
 												id={b?.id}
 												{total}
 												{min}
-												concat
-												started_at={b.started_at}
-												len={b.started_at ? (b?.duration_ms ?? now - b?.started_at) : 0}
-												running={b?.duration_ms == undefined}
+												gray
+												started_at={b.created_at}
+												len={waitingLen < 100 ? 0 : waitingLen - 100}
+												running={b?.started_at == undefined}
 											/>
-										{/if}
-									</div>
-								{/each}
-							</div>
+											{#if b.started_at}
+												<TimelineBar
+													position={waitingLen < 100 ? 'center' : 'right'}
+													id={b?.id}
+													{total}
+													{min}
+													concat
+													started_at={b.started_at}
+													len={b.started_at ? (b?.duration_ms ?? now - b?.started_at) : 0}
+													running={b?.duration_ms == undefined}
+												/>
+											{/if}
+										</div>
+									{:else}
+										<div class="flex w-full p-1 pb-2 pl-12">
+											<div class="text-xs text-secondary">
+												<!-- Waiting for executor/Suspend {JSON.stringify(b)} -->
+											</div>
+										</div>
+									{/if}
+								{/snippet}
+							</VirtualList>
 						{/if}</div
 					></div
 				>

frontend/src/lib/components/FlowTimelineBar.svelte

@@ -21,8 +21,6 @@
 		showZoomButtons?: boolean
 		onZoom?: () => void
 		zoom?: 'in' | 'out'
-		hasMoreIterations?: boolean
-		loadPreviousIterations?: () => void
 		onSelectIteration?: (id: string) => void
 		idToIterationIndex?: (id: string) => number | undefined
 		showIterations?: string[]
@@ -39,8 +37,6 @@
 		showZoomButtons = false,
 		onZoom,
 		zoom = 'in',
-		hasMoreIterations,
-		loadPreviousIterations,
 		onSelectIteration,
 		idToIterationIndex,
 		showIterations,
@@ -112,26 +108,61 @@
 		return { left: leftPercent, width: widthPercent }
 	}
 
-	function getOverlapOpacity(item: TimelineItem, allItems: TimelineItem[]): number {
-		if (!item.started_at) return 1
+	// More efficient version using sweep line algorithm for computing all overlaps at once
+	function computeAllOverlaps(items: TimelineItem[]): Record<string, number> {
+		const overlapCounts = new Map<string, number>()
 
-		const itemEnd = item.duration_ms ? item.started_at + item.duration_ms : now
-		let overlapCount = 0
-
-		for (const otherItem of allItems) {
-			if (otherItem.id === item.id || !otherItem.started_at) continue
-
-			const otherEnd = otherItem.duration_ms ? otherItem.started_at + otherItem.duration_ms : now
-
-			// Check if time ranges overlap
-			if (item.started_at < otherEnd && otherItem.started_at < itemEnd) {
-				overlapCount++
-			}
+		// Create events for start and end times
+		interface Event {
+			time: number
+			type: 'start' | 'end'
+			itemId: string
 		}
 
-		// Base opacity of 1, reduce by 0.2 for each overlap, minimum 0.3
-		return Math.max(0.3, 1 - overlapCount * 0.2)
+		const events: Event[] = []
+
+		for (const item of items) {
+			if (!item.started_at) continue
+
+			const endTime = item.duration_ms ? item.started_at + item.duration_ms : now
+			events.push({ time: item.started_at, type: 'start', itemId: item.id })
+			events.push({ time: endTime, type: 'end', itemId: item.id })
+			overlapCounts.set(item.id, 0)
+		}
+
+		// Sort events by time, with end events before start events at the same time
+		events.sort((a, b) => {
+			if (a.time !== b.time) return a.time - b.time
+			return a.type === 'end' ? -1 : 1
+		})
+
+		// Sweep through events
+		const activeItems = new Set<string>()
+
+		for (const event of events) {
+			if (event.type === 'start') {
+				// Count current active items as overlaps for this item
+				overlapCounts.set(event.itemId, activeItems.size)
+
+				// Update overlap counts for all currently active items
+				for (const activeId of activeItems) {
+					overlapCounts.set(activeId, overlapCounts.get(activeId)! + 1)
+				}
+
+				activeItems.add(event.itemId)
+			} else {
+				activeItems.delete(event.itemId)
+			}
+		}
+		return Object.fromEntries(overlapCounts.entries())
 	}
+
+	// At component level, compute once when items change
+	const allOverlaps = $derived(computeAllOverlaps(filteredItems))
+	const maximumOverlaps = $derived(Math.max(...Object.values(allOverlaps)))
+	const opacity = $derived(Math.max(0.02, 1 / maximumOverlaps))
+	// Then in your template, use:
+	// allOverlaps.get(item.id) ?? 0
 </script>
 
 {#if min && filteredItems.length > 0 && startItem?.started_at}
@@ -155,24 +186,6 @@
 					{/if}
 				</button>
 			</div>
-		{:else if hasMoreIterations}
-			<Tooltip
-				class="hover:text-primary hover:bg-surface p-1 -my-1 w-24 rounded-md flex items-center justify-center"
-				openDelay={100}
-			>
-				<button
-					class="text-2xs text-primary whitespace-nowrap"
-					onclick={(e) => {
-						e.stopPropagation()
-						loadPreviousIterations?.()
-					}}
-				>
-					load more...
-				</button>
-				{#snippet text()}
-					Load previous iterations
-				{/snippet}
-			</Tooltip>
 		{:else}
 			<div class="w-24"></div>
 		{/if}
@@ -205,7 +218,8 @@
 				{#each filteredItems as item, i}
 					{#if item.started_at}
 						{@const position = calculateItemPosition(item)}
-						{@const opacity = getOverlapOpacity(item, filteredItems)}
+						<!-- {overlapCount}
+						{opacity} -->
 						<Tooltip
 							style="left: {position.left}%; width: {position.width}%"
 							class="h-full absolute top-0"

frontend/src/lib/components/QueuePosition.svelte

@@ -27,9 +27,13 @@
 					JobService.getScheduledFor({
 						workspace: workspace,
 						id: jobId
-					}).then((response) => {
-						scheduledFor = response
 					})
+						.then((response) => {
+							scheduledFor = response
+						})
+						.catch((error) => {
+							console.error('Failed to fetch scheduled for:', error)
+						})
 				} catch (error) {
 					console.error('Failed to fetch scheduled for:', error)
 				}

frontend/src/lib/components/flows/flowModuleNextId.ts

@@ -6,6 +6,8 @@ import { charsToNumber, numberToChars } from './idUtils'
 // Computes the next available id
 export function nextId(flowState: FlowState, fullFlow: OpenFlow): string {
 	const allIds = dfs(fullFlow.value.modules, (fm) => fm.id)
+	console.log('allIds', allIds)
+
 	const max = allIds.concat(Object.keys(flowState)).reduce((acc, key) => {
 		if (key.length >= 4) {
 			return acc

frontend/src/lib/components/graph/model.ts

@@ -27,11 +27,6 @@ export type DurationStatus = {
 	byJob: Record<string, { created_at?: number; started_at?: number; duration_ms?: number }>
 }
 
-export type GlobalIterationBounds = {
-	iteration_from?: number
-	iteration_total?: number
-}
-
 export type FlowStatusViewerContext = {
 	flowStateStore?: FlowState
 	retryStatus: StateStore<Record<string, number | undefined>>
@@ -58,6 +53,10 @@ export type GraphModuleState = {
 	selectedForloopIndex?: number
 	selectedForLoopSetManually?: boolean
 	flow_jobs_success?: (boolean | undefined)[]
+	flow_jobs_duration?: {
+		started_at?: (string | undefined)[]
+		duration_ms?: (number | undefined)[]
+	}
 	flow_jobs?: string[]
 	iteration_total?: number
 	retries?: number

openflow.openapi.yaml

@@ -89,7 +89,7 @@ components:
               minimum: 0
               maximum: 100
         retry_if:
-          $ref: '#/components/schemas/RetryIf'
+          $ref: "#/components/schemas/RetryIf"
 
     RetryIf:
       type: object
@@ -569,6 +569,17 @@ components:
           type: array
           items:
             type: boolean
+        flow_jobs_duration:
+          type: object
+          properties:
+            started_at:
+              type: array
+              items:
+                type: string
+            duration_ms:
+              type: array
+              items:
+                type: integer
         branch_chosen:
           type: object
           properties: