feat: trigger scripts and have flows being triggered by checking new external events regularly (#200)

* v1 trigger scripts * progress * stop early condition * backend execution model * progress * progress * progress * works but todo: collect result, render forloop, convert forloopraw to seq for frontend * collect result * v1 * that's enough * sed
2022-07-17 13:17:45 +02:00
parent 080207318e
commit 738d429f2f
34 changed files with 1367 additions and 675 deletions
--- a/backend/migrations/20220713142758_script_trigger.down.sql
+++ b/backend/migrations/20220713142758_script_trigger.down.sql
@@ -0,0 +1 @@
+-- Add down migration script here
--- a/backend/migrations/20220713142758_script_trigger.up.sql
+++ b/backend/migrations/20220713142758_script_trigger.up.sql
@@ -0,0 +1,3 @@
+-- Add up migration script here
+ALTER TABLE script ADD COLUMN trigger_reco_interval INTEGER;
+ALTER TABLE completed_job ADD COLUMN is_skipped BOOLEAN NOT NULL DEFAULT FALSE;
--- a/backend/openapi.yaml
+++ b/backend/openapi.yaml
@@ -1602,7 +1602,8 @@ paths:
                language:
                  type: string
                  enum: [python3, deno]
-
+                trigger_reco_interval:
+                  type: integer
              required:
                - path
                - summary
@@ -2971,6 +2972,8 @@ components:
        language:
          type: string
          enum: [python3, deno]
+        trigger_reco_interval:
+          type: integer
      required:
        - hash
        - path
@@ -3033,7 +3036,15 @@ components:
          format: date-time
        job_kind:
          type: string
-          enum: ["script", "preview", "dependencies", "flow", "flowpreview"]
+          enum:
+            [
+              "script",
+              "preview",
+              "dependencies",
+              "flow",
+              "flowpreview",
+              "script_hub",
+            ]
        schedule_path:
          type: string
        permissioned_as:
@@ -3103,7 +3114,15 @@ components:
          type: string
        job_kind:
          type: string
-          enum: ["script", "preview", "dependencies", "flow", "flowpreview"]
+          enum:
+            [
+              "script",
+              "preview",
+              "dependencies",
+              "flow",
+              "flowpreview",
+              "script_hub",
+            ]
        schedule_path:
          type: string
        permissioned_as:
@@ -3120,6 +3139,8 @@ components:
        language:
          type: string
          enum: [python3, deno]
+        is_skipped:
+          type: boolean
      required:
        - id
        - created_by
@@ -3131,6 +3152,7 @@ components:
        - job_kind
        - permissioned_as
        - is_flow_step
+        - is_skipped

    Job:
      allOf:
@@ -3754,6 +3776,10 @@ components:
            $ref: "#/components/schemas/InputTransform"
        value:
          $ref: "#/components/schemas/FlowModuleValue"
+        stop_after_if_expr:
+          type: string
+        skip_if_stopped:
+          type: boolean
      required:
        - input_transform
        - value
@@ -3775,6 +3801,10 @@ components:
    FlowModuleValue:
      type: object
      properties:
+        value:
+          $ref: "#/components/schemas/FlowValue"
+        iterator:
+          $ref: "#/components/schemas/InputTransform"
        path:
          type: string
        content:
@@ -3790,6 +3820,7 @@ components:
            - script
            - flow
            - rawscript
+            - forloopflow
      required:
        - type

@@ -3841,7 +3872,19 @@ components:
          format: uuid
        event:
          type: string
-
+        iterator:
+          type: object
+          properties:
+            index:
+              type: integer
+            itered:
+              type: array
+              items: {}
+            args: {}
+        forloop_jobs:
+          type: array
+          items:
+            type: string
      required: [type]

    SlackToken:
--- a/backend/sqlx-data.json
+++ b/backend/sqlx-data.json
@@ -2095,61 +2095,6 @@
      ]
    }
  },
-  "a8507084c0f45f4c08c0b317d69db26f97295bb2d410e3e5b8da2cd420536145": {
-    "query": "INSERT INTO completed_job as cj\n            (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, args, result, logs, \n            raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step)\n            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING id",
-    "describe": {
-      "columns": [
-        {
-          "ordinal": 0,
-          "name": "id",
-          "type_info": "Uuid"
-        }
-      ],
-      "parameters": {
-        "Left": [
-          "Varchar",
-          "Uuid",
-          "Uuid",
-          "Varchar",
-          "Timestamptz",
-          "Int4",
-          "Bool",
-          "Int8",
-          "Varchar",
-          "Jsonb",
-          "Jsonb",
-          "Text",
-          "Text",
-          "Bool",
-          "Varchar",
-          "Text",
-          {
-            "Custom": {
-              "name": "job_kind",
-              "kind": {
-                "Enum": [
-                  "script",
-                  "preview",
-                  "flow",
-                  "dependencies",
-                  "flowpreview",
-                  "script_hub"
-                ]
-              }
-            }
-          },
-          "Varchar",
-          "Varchar",
-          "Jsonb",
-          "Jsonb",
-          "Bool"
-        ]
-      },
-      "nullable": [
-        false
-      ]
-    }
-  },
  "a98b2d68f023f46ab91167d3147416df672c2aed2ba5ab70e98a9da5fa47255a": {
    "query": "INSERT INTO workspace_settings\n            (workspace_id)\n            VALUES ($1)",
    "describe": {
@@ -2297,6 +2242,62 @@
      "nullable": []
    }
  },
+  "b032ac8a438eb36a61bf5a81a0c41eb10646349e8c79014f83142be35139205c": {
+    "query": "INSERT INTO completed_job as cj\n            (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, args, result, logs, \n            raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, is_skipped)\n            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING id",
+    "describe": {
+      "columns": [
+        {
+          "ordinal": 0,
+          "name": "id",
+          "type_info": "Uuid"
+        }
+      ],
+      "parameters": {
+        "Left": [
+          "Varchar",
+          "Uuid",
+          "Uuid",
+          "Varchar",
+          "Timestamptz",
+          "Int4",
+          "Bool",
+          "Int8",
+          "Varchar",
+          "Jsonb",
+          "Jsonb",
+          "Text",
+          "Text",
+          "Bool",
+          "Varchar",
+          "Text",
+          {
+            "Custom": {
+              "name": "job_kind",
+              "kind": {
+                "Enum": [
+                  "script",
+                  "preview",
+                  "flow",
+                  "dependencies",
+                  "flowpreview",
+                  "script_hub"
+                ]
+              }
+            }
+          },
+          "Varchar",
+          "Varchar",
+          "Jsonb",
+          "Jsonb",
+          "Bool",
+          "Bool"
+        ]
+      },
+      "nullable": [
+        false
+      ]
+    }
+  },
  "b20977e70ebac7ccbaec5a2a1e940301dd331a5f9a4be67a27cfbff8619ac8f0": {
    "query": "INSERT INTO usr\n            (workspace_id, email, username, is_admin)\n            VALUES ($1, $2, $3, true)",
    "describe": {
@@ -2419,40 +2420,6 @@
      ]
    }
  },
-  "be33c6eb702c149044650d49b3c50493d7538d590be3f4ff6242fea85c57c667": {
-    "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13)",
-    "describe": {
-      "columns": [],
-      "parameters": {
-        "Left": [
-          "Varchar",
-          "Int8",
-          "Varchar",
-          "Int8Array",
-          "Text",
-          "Text",
-          "Text",
-          "Varchar",
-          "Text",
-          "Bool",
-          "Jsonb",
-          "Text",
-          {
-            "Custom": {
-              "name": "script_lang",
-              "kind": {
-                "Enum": [
-                  "python3",
-                  "deno"
-                ]
-              }
-            }
-          }
-        ]
-      },
-      "nullable": []
-    }
-  },
  "bf1d8e043338867e1da1ed236ff6c85a566d5fd58d4b0d5c3a10454513811ba3": {
    "query": "UPDATE workspace_settings\n            SET slack_team_id = null, slack_name = null WHERE workspace_id = $1",
    "describe": {
@@ -3012,6 +2979,41 @@
      "nullable": []
    }
  },
+  "f503c8f37507c89bf99063f49ef72e2d021898583708e776fc58d99809f4bc3c": {
+    "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, trigger_reco_interval) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)",
+    "describe": {
+      "columns": [],
+      "parameters": {
+        "Left": [
+          "Varchar",
+          "Int8",
+          "Varchar",
+          "Int8Array",
+          "Text",
+          "Text",
+          "Text",
+          "Varchar",
+          "Text",
+          "Bool",
+          "Jsonb",
+          "Text",
+          {
+            "Custom": {
+              "name": "script_lang",
+              "kind": {
+                "Enum": [
+                  "python3",
+                  "deno"
+                ]
+              }
+            }
+          },
+          "Int4"
+        ]
+      },
+      "nullable": []
+    }
+  },
  "f7906298e4204ad55ec84021bb2461f369386493519637279f1188227230c580": {
    "query": "SELECT lock, lock_error_logs FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')",
    "describe": {
--- a/backend/src/flow.rs
+++ b/backend/src/flow.rs
@@ -60,19 +60,21 @@ pub struct NewFlow {
    pub schema: Option<Schema>,
 }

-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct FlowValue {
    pub modules: Vec<FlowModule>,
    pub failure_module: Option<FlowModule>,
 }

-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct FlowModule {
    pub input_transform: HashMap<String, InputTransform>,
    pub value: FlowModuleValue,
+    pub stop_after_if_expr: Option<String>,
+    pub skip_if_stopped: Option<bool>,
 }

-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(
    tag = "type",
    rename_all(serialize = "lowercase", deserialize = "lowercase")
@@ -80,17 +82,24 @@ pub struct FlowModule {
 pub enum InputTransform {
    Static { value: serde_json::Value },
    Javascript { expr: String },
-    Resource { path: String },
 }

-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(
    tag = "type",
    rename_all(serialize = "lowercase", deserialize = "lowercase")
 )]
 pub enum FlowModuleValue {
-    Script { path: String },
-    Flow { path: String },
+    Script {
+        path: String,
+    },
+    ForloopFlow {
+        iterator: InputTransform,
+        value: Box<FlowValue>,
+    },
+    Flow {
+        path: String,
+    },
    RawScript(RawCode),
 }

@@ -315,6 +324,8 @@ mod tests {
                    value: FlowModuleValue::Script {
                        path: "test".to_string(),
                    },
+                    stop_after_if_expr: None,
+                    skip_if_stopped: Some(false),
                },
                FlowModule {
                    input_transform: HashMap::new(),
@@ -323,6 +334,28 @@ mod tests {
                        language: crate::scripts::ScriptLang::Deno,
                        path: None,
                    }),
+                    stop_after_if_expr: Some("foo = 'bar'".to_string()),
+                    skip_if_stopped: None,
+                },
+                FlowModule {
+                    input_transform: [(
+                        "iterand".to_string(),
+                        InputTransform::Static {
+                            value: serde_json::json!(vec![1, 2, 3]),
+                        },
+                    )]
+                    .into(),
+                    value: FlowModuleValue::ForloopFlow {
+                        iterator: InputTransform::Static {
+                            value: serde_json::json!([1, 2, 3]),
+                        },
+                        value: Box::new(FlowValue {
+                            modules: vec![],
+                            failure_module: None,
+                        }),
+                    },
+                    stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
+                    skip_if_stopped: None,
                },
            ],
            failure_module: Some(FlowModule {
@@ -330,6 +363,8 @@ mod tests {
                value: FlowModuleValue::Flow {
                    path: "test".to_string(),
                },
+                stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
+                skip_if_stopped: None,
            }),
        };
        println!("{}", serde_json::json!(fv).to_string());
--- a/backend/src/jobs.rs
+++ b/backend/src/jobs.rs
@@ -10,16 +10,16 @@ use chrono::Duration;
 use sql_builder::prelude::*;
 use sqlx::{query_scalar, Postgres, Transaction};
 use std::collections::HashMap;
+use tracing::instrument;

-use crate::js_eval::eval_timeout;
 use crate::scripts::{get_hub_script_by_path, ScriptLang};
-use crate::users::create_token_for_owner;
+use crate::worker_flow::init_flow_status;
 use crate::{
    audit::{audit_log, ActionKind},
    db::{UserDB, DB},
    error,
    error::Error,
-    flow::{FlowModuleValue, FlowValue, InputTransform},
+    flow::FlowValue,
    schedule::get_schedule_opt,
    scripts::ScriptHash,
    users::{owner_to_token_owner, Authed},
@@ -32,7 +32,7 @@ use axum::{
 };
 use hyper::StatusCode;
 use serde::{Deserialize, Serialize};
-use serde_json::{json, Map, Value};
+use serde_json::{Map, Value};
 use sql_builder::SqlBuilder;

 use ulid::Ulid;
@@ -114,6 +114,7 @@ struct CompletedJob {
    raw_flow: Option<serde_json::Value>,
    is_flow_step: bool,
    language: Option<ScriptLang>,
+    is_skipped: bool,
 }

 #[derive(Deserialize, Clone, Copy)]
@@ -185,7 +186,7 @@ pub async fn run_job_by_path(
    Ok((StatusCode::CREATED, uuid.to_string()))
 }

-async fn script_path_to_payload<'c>(
+pub async fn script_path_to_payload<'c>(
    script_path: &str,
    db: &mut Transaction<'c, Postgres>,
    w_id: &String,
@@ -438,6 +439,7 @@ async fn list_jobs(
            "flow_status",
            "is_flow_step",
            "language",
+            "false as is_skipped",
        ],
    );
    let sqlc = list_completed_jobs_query(
@@ -472,6 +474,7 @@ async fn list_jobs(
            "flow_status",
            "is_flow_step",
            "language",
+            "is_skipped",
        ],
    );
    let sql = format!(
@@ -804,7 +807,7 @@ async fn get_job_from_id<'c>(
    Ok((job_option, tx))
 }

-async fn get_queued_job<'c>(
+pub async fn get_queued_job<'c>(
    id: Uuid,
    w_id: &str,
    tx: &mut Transaction<'c, Postgres>,
@@ -840,24 +843,6 @@ pub enum JobKind {
    FlowPreview,
 }

-#[derive(Serialize, Deserialize, Debug)]
-pub struct FlowStatus {
-    pub step: i32,
-    pub modules: Vec<FlowStatusModule>,
-    pub failure_module: FlowStatusModule,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(tag = "type")]
-pub enum FlowStatusModule {
-    WaitingForPriorSteps,
-    WaitingForEvent { event: String },
-    WaitingForExecutor { job: Uuid },
-    InProgress { job: Uuid },
-    Success { job: Uuid },
-    Failure { job: Uuid },
-}
-
 #[derive(sqlx::FromRow)]
 struct UnifiedJob {
    workspace_id: String,
@@ -883,6 +868,7 @@ struct UnifiedJob {
    flow_status: Option<serde_json::Value>,
    is_flow_step: bool,
    language: Option<ScriptLang>,
+    is_skipped: bool,
 }

 impl From<UnifiedJob> for Job {
@@ -914,6 +900,7 @@ impl From<UnifiedJob> for Job {
                raw_flow: None,
                is_flow_step: uj.is_flow_step,
                language: uj.language,
+                is_skipped: uj.is_skipped,
            }),
            "QueuedJob" => Job::QueuedJob(QueuedJob {
                workspace_id: uj.workspace_id,
@@ -972,6 +959,7 @@ struct PreviewFlow {
    args: Option<Map<String, Value>>,
 }

+#[derive(Debug)]
 pub enum JobPayload {
    ScriptHub {
        path: String,
@@ -992,6 +980,7 @@ pub enum JobPayload {
    },
 }

+#[instrument(level = "trace", skip_all)]
 pub async fn push<'c>(
    mut tx: Transaction<'c, Postgres>,
    workspace_id: &str,
@@ -1125,13 +1114,7 @@ pub async fn push<'c>(
        }
    };

-    let flow_status = raw_flow.as_ref().map(|f| FlowStatus {
-        step: 0,
-        modules: (0..f.modules.len())
-            .map(|_| FlowStatusModule::WaitingForPriorSteps)
-            .collect(),
-        failure_module: FlowStatusModule::WaitingForPriorSteps,
-    });
+    let flow_status = raw_flow.as_ref().map(init_flow_status);
    let uuid = sqlx::query_scalar!(
        "INSERT INTO queue
            (workspace_id, id, parent_job, created_by, permissioned_as, scheduled_for, 
@@ -1155,7 +1138,9 @@ pub async fn push<'c>(
        language: ScriptLang
    )
    .fetch_one(&mut tx)
-    .await?;
+    .await
+    .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?;
+
    let uuid_string = job_id.to_string();
    let uuid_str = uuid_string.as_str();
    let mut hm = HashMap::from([("uuid", uuid_str), ("permissioned_as", &permissioned_as)]);
@@ -1192,7 +1177,8 @@ pub async fn push<'c>(
    Ok((uuid, tx))
 }

-pub async fn add_completed_job_error<E: ToString>(
+#[instrument(level = "trace", skip_all)]
+pub async fn add_completed_job_error<E: ToString + std::fmt::Debug>(
    db: &DB,
    queued_job: &QueuedJob,
    logs: String,
@@ -1207,6 +1193,7 @@ pub async fn add_completed_job_error<E: ToString>(
        db,
        &queued_job,
        false,
+        false,
        Some(output_map.clone()),
        format!("{}\n{}", logs, e.to_string()),
    )
@@ -1214,14 +1201,17 @@ pub async fn add_completed_job_error<E: ToString>(
    Ok((a, output_map))
 }

+#[instrument(level = "trace", skip_all)]
 pub async fn add_completed_job(
    db: &DB,
    queued_job: &QueuedJob,
    success: bool,
+    skipped: bool,
    result: Option<Map<String, Value>>,
    logs: String,
 ) -> Result<Uuid, Error> {
    let result_json = result.map(serde_json::Value::Object);
+    let job_id = queued_job.id.clone();
    let duration = (chrono::Utc::now() - queued_job.started_at.unwrap_or(queued_job.created_at))
        .num_seconds() as i32;
    let _ = sqlx::query!(
@@ -1229,8 +1219,8 @@ pub async fn add_completed_job(
            (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, \
        args, result, logs, 
            raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, \
-            is_flow_step)
-            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) \
+            is_flow_step, is_skipped)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) \
        ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) \
        RETURNING id",
        queued_job.workspace_id,
@@ -1254,150 +1244,33 @@ pub async fn add_completed_job(
        queued_job.permissioned_as,
        queued_job.flow_status,
        queued_job.raw_flow,
-        queued_job.is_flow_step
+        queued_job.is_flow_step,
+        skipped
    )
    .fetch_one(db)
-    .await?;
+    .await
+    .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?;
    tracing::debug!("Added completed job {}", queued_job.id);
    Ok(queued_job.id)
 }

-pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<i32> {
-    let r = sqlx::query_scalar!(
-        "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1",
-        id
-    )
-    .fetch_one(db)
-    .await?
-    .ok_or_else(|| Error::InternalErr(format!("not found step")))?;
-    Ok(r)
-}
-pub async fn update_flow_status_in_progress(
-    db: &DB,
-    w_id: &str,
-    flow: Uuid,
-    job_in_progress: Uuid,
-) -> error::Result<()> {
-    let step = get_step_of_flow_status(db, flow).await?;
-    sqlx::query(&format!(
-        "UPDATE queue
-            SET flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1)
-            WHERE id = $2 AND workspace_id = $3",
-        step
-    ))
-    .bind(serde_json::json!(FlowStatusModule::InProgress {
-        job: job_in_progress
-    }))
-    .bind(flow)
-    .bind(w_id)
-    .execute(db)
-    .await?;
-    Ok(())
-}
-
-pub async fn update_flow_status_after_job_completion(
-    db: &DB,
-    job: &QueuedJob,
-    success: bool,
-    result: Option<Map<String, Value>>,
-) -> error::Result<()> {
-    tracing::info!("HANDLE FLOW: {job:?} {success} {result:?}");
-
-    let mut tx = db.begin().await?;
-
-    let w_id = &job.workspace_id;
-
-    let flow = job
-        .parent_job
-        .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?;
-
-    let old_status_json = sqlx::query_scalar!(
-        "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2",
-        flow,
-        w_id
-    )
-    .fetch_one(&mut tx)
-    .await?
-    .ok_or_else(|| Error::InternalErr(format!("requiring a previous status")))?;
-
-    let old_status = serde_json::from_value::<FlowStatus>(old_status_json)
-        .ok()
-        .ok_or_else(|| {
-            Error::InternalErr(format!("requiring status to be parsabled as FlowStatus"))
-        })?;
-
-    let last_step = (old_status.step + 1) as usize == old_status.modules.len();
-    let new_status = if success {
-        FlowStatusModule::Success { job: job.id }
-    } else {
-        FlowStatusModule::Failure { job: job.id }
-    };
-
-    sqlx::query(&format!(
-        "UPDATE queue
-            SET 
-                flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {}}}', $1), '{{\"step\"}}', $2)
-            WHERE id = $3",
-        old_status.step,
-    ))
-    .bind(serde_json::json!(new_status))
-    .bind(serde_json::json!(old_status.step + 1))
-    .bind(flow)
-    .execute(&mut tx)
-    .await?;
-
-    tracing::info!("UPDATE: {:?}", new_status);
-
-    let flow_job = get_queued_job(flow, w_id, &mut tx)
-        .await?
-        .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
-    tx.commit().await?;
-
-    let done = if !success || last_step {
-        add_completed_job(
-            db,
-            &flow_job,
-            success,
-            result,
-            "Flow job completed".to_string(),
-        )
-        .await?;
-        true
-    } else {
-        if let Err(err) = handle_flow(&flow_job, db, result).await {
-            let _ = add_completed_job_error(
-                db,
-                &flow_job,
-                "Unexpected error during flow chaining:\n".to_string(),
-                err,
-            )
-            .await;
-            true
-        } else {
-            false
-        }
-    };
-
-    if done {
-        postprocess_queued_job(flow_job.schedule_path, &w_id, flow, db).await?;
-    }
-
-    Ok(())
-}
-
+#[instrument(level = "trace", skip_all)]
 pub async fn postprocess_queued_job(
    schedule_path: Option<String>,
+    script_path: Option<String>,
    w_id: &str,
    job_id: Uuid,
    db: &DB,
 ) -> crate::error::Result<()> {
    let _ = delete_job(db, w_id, job_id).await?;
-    schedule_again_if_scheduled(schedule_path, &w_id, db).await?;
+    schedule_again_if_scheduled(schedule_path, script_path, &w_id, db).await?;
    Ok(())
 }

+#[instrument(level = "trace", skip_all)]
 pub async fn schedule_again_if_scheduled(
    schedule_path: Option<String>,
+    script_path: Option<String>,
    w_id: &str,
    db: &DB,
 ) -> crate::error::Result<()> {
@@ -1406,7 +1279,8 @@ pub async fn schedule_again_if_scheduled(
        let schedule = get_schedule_opt(&mut tx, &w_id, &schedule_path)
            .await?
            .unwrap();
-        if schedule.enabled {
+        if schedule.enabled && script_path.is_some() && script_path.unwrap() == schedule.script_path
+        {
            tx = crate::schedule::push_scheduled_job(tx, schedule).await?;
        }
        tx.commit().await?;
@@ -1414,167 +1288,6 @@ pub async fn schedule_again_if_scheduled(
    Ok(())
 }

-pub async fn handle_flow(
-    job: &QueuedJob,
-    db: &sqlx::Pool<sqlx::Postgres>,
-    last_result: Option<Map<String, serde_json::Value>>,
-) -> anyhow::Result<()> {
-    let value = job
-        .raw_flow
-        .as_ref()
-        .ok_or_else(|| Error::InternalErr(format!("requiring a raw flow value")))?
-        .to_owned();
-    let flow = serde_json::from_value::<FlowValue>(value.to_owned())?;
-    push_next_flow_job(job, flow, db, last_result).await?;
-    Ok(())
-}
-
-async fn transform_input(
-    flow_args: &Option<serde_json::Value>,
-    last_result: Option<Map<String, serde_json::Value>>,
-    input_transform: &HashMap<String, InputTransform>,
-    workspace: &str,
-    token: &str,
-    steps: Vec<String>,
-) -> anyhow::Result<Option<Map<String, serde_json::Value>>> {
-    let mut mapped = serde_json::Map::new();
-
-    for (key, val) in input_transform.into_iter() {
-        match val {
-            InputTransform::Static { value } => {
-                mapped.insert(key.to_string(), value.to_owned());
-                ()
-            }
-            _ => (),
-        };
-    }
-
-    for (key, val) in input_transform.into_iter() {
-        match val {
-            InputTransform::Static { value: _ } => (),
-            InputTransform::Javascript { expr } => {
-                let previous_result =
-                    serde_json::Value::Object(last_result.clone().unwrap_or_else(|| Map::new()));
-                let flow_input = flow_args.clone().unwrap_or_else(|| json!({}));
-                let v = eval_timeout(
-                    expr.to_string(),
-                    vec![
-                        ("params".to_string(), serde_json::json!(mapped)),
-                        ("previous_result".to_string(), previous_result),
-                        ("flow_input".to_string(), flow_input),
-                    ],
-                    workspace,
-                    token,
-                    steps.clone(),
-                )
-                .await
-                .map_err(|e| {
-                    Error::ExecutionErr(format!(
-                        "Error during isolated evaluation of expression `{expr}`:\n{e}"
-                    ))
-                })?;
-                mapped.insert(key.to_string(), v);
-                ()
-            }
-            _ => Err(error::Error::BadRequest(format!(
-                "impossible to handle unknown input transform"
-            )))?,
-        }
-    }
-
-    Ok(Some(mapped))
-}
-
-async fn push_next_flow_job(
-    job: &QueuedJob,
-    flow: FlowValue,
-    db: &sqlx::Pool<sqlx::Postgres>,
-    last_result: Option<Map<String, serde_json::Value>>,
-) -> anyhow::Result<()> {
-    let flow_status_json = job
-        .flow_status
-        .as_ref()
-        .ok_or_else(|| Error::InternalErr(format!("not found status for flow job {:?}", job.id)))?;
-    let status = serde_json::from_value::<FlowStatus>(flow_status_json.to_owned())?;
-    let i = status.step as usize;
-
-    if flow.modules.len() > i {
-        let module = &flow.modules[i];
-        let mut tx = db.begin().await?;
-        let job_payload = match &module.value {
-            FlowModuleValue::Script { path: script_path } => {
-                script_path_to_payload(script_path, &mut tx, &job.workspace_id).await?
-            }
-            FlowModuleValue::RawScript(raw_code) => JobPayload::Code(raw_code.clone()),
-            a @ _ => {
-                tracing::info!("Unrecognized module values {:?}", a);
-                Err(Error::BadRequest(format!(
-                    "Unrecognized module values {:?}",
-                    a
-                )))?
-            }
-        };
-
-        let token = create_token_for_owner(
-            &db,
-            &job.workspace_id,
-            &job.permissioned_as,
-            crate::users::NewToken {
-                label: Some("transform-input".to_string()),
-                expiration: Some(chrono::Utc::now() + chrono::Duration::seconds(10)),
-            },
-            &job.created_by,
-        )
-        .await?;
-
-        let args = transform_input(
-            &job.args,
-            last_result,
-            &module.input_transform,
-            &job.workspace_id,
-            &token,
-            status
-                .modules
-                .into_iter()
-                .map(|x| match x {
-                    FlowStatusModule::Success { job } => job.to_string(),
-                    _ => "invalid step status".to_string(),
-                })
-                .collect(),
-        )
-        .await?; //job.args
-        let (uuid, mut tx) = push(
-            tx,
-            &job.workspace_id,
-            job_payload,
-            args,
-            &job.created_by,
-            job.permissioned_as.to_owned(),
-            None,
-            None,
-            Some(job.id),
-            true,
-        )
-        .await?;
-
-        sqlx::query(&format!(
-            "UPDATE queue
-            SET 
-                flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1)
-            WHERE id = $2",
-            i
-        ))
-        .bind(serde_json::json!(FlowStatusModule::WaitingForExecutor {
-            job: uuid
-        }))
-        .bind(job.parent_job)
-        .execute(&mut tx)
-        .await?;
-        tx.commit().await?;
-    }
-    Ok(())
-}
-
 pub async fn pull(db: &DB) -> Result<Option<QueuedJob>, crate::Error> {
    let now = chrono::Utc::now();

@@ -1599,6 +1312,7 @@ pub async fn pull(db: &DB) -> Result<Option<QueuedJob>, crate::Error> {
    Ok(job)
 }

+#[instrument(level = "trace", skip_all)]
 pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate::Error> {
    let job_removed = sqlx::query_scalar!(
        "DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
@@ -1606,7 +1320,8 @@ pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate::
        job_id
    )
    .fetch_one(db)
-    .await?
+    .await
+    .map_err(|e| Error::InternalErr(format!("Error during deletion of job {job_id}: {e}")))?
    .unwrap_or(0)
        == 1;
    tracing::debug!("Job {job_id} deletion was achieved with success: {job_removed}");
--- a/backend/src/js_eval.rs
+++ b/backend/src/js_eval.rs
@@ -21,27 +21,33 @@ use tokio::time::timeout;
 use crate::client;
 use crate::error::Error;

+pub struct EvalCreds {
+    pub workspace: String,
+    pub token: String,
+}
+
 pub async fn eval_timeout(
    expr: String,
    env: Vec<(String, serde_json::Value)>,
-    workspace: &str,
-    token: &str,
+    creds: Option<EvalCreds>,
    steps: Vec<String>,
 ) -> anyhow::Result<serde_json::Value> {
    let expr2 = expr.clone();
    let (sender, mut receiver) = oneshot::channel::<IsolateHandle>();
-    let (workspace, token) = (workspace.to_string().clone(), token.to_string().clone());
    timeout(
        std::time::Duration::from_millis(2000),
        tokio::task::spawn_blocking(move || {
+            let mut ops = vec![];

-            let mut ops = vec![
-                // An op for summing an array of numbers
-                // The op-layer automatically deserializes inputs
-                // and serializes the returned Result & value
-                op_variable::decl(),
-                op_resource::decl(),
-            ];
+            if creds.is_some() {
+                ops.extend([
+                    // An op for summing an array of numbers
+                    // The op-layer automatically deserializes inputs
+                    // and serializes the returned Result & value
+                    op_variable::decl(),
+                    op_resource::decl(),
+                ])
+            }

            if !steps.is_empty() {
                ops.push(op_get_result::decl())
@@ -72,8 +78,7 @@ pub async fn eval_timeout(
                .into_iter()
                .fold(expr, replace_with_await);

-            let r =
-                runtime.block_on(eval(&mut js_runtime, &expr, env, &workspace, &token, steps))?;
+            let r = runtime.block_on(eval(&mut js_runtime, &expr, env, creds, steps))?;

            Ok(r) as anyhow::Result<Value>
        }),
@@ -123,8 +128,7 @@ async fn eval(
    context: &mut JsRuntime,
    expr: &str,
    env: Vec<(String, serde_json::Value)>,
-    workspace: &str,
-    token: &str,
+    creds: Option<EvalCreds>,
    steps: Vec<String>,
 ) -> anyhow::Result<serde_json::Value> {
    let expr = expr.trim();
@@ -135,9 +139,10 @@ async fn eval(
            .join("\n"),
        expr.split(SPLIT_PAT).last().unwrap_or_else(|| "")
    );
-    let steps_code = if !steps.is_empty() {
-        format!(
-            r#"
+    let (steps_code, api_code) = if let Some(EvalCreds { workspace, token }) = creds {
+        let steps_code = if !steps.is_empty() {
+            format!(
+                r#"
 let steps = [{}];
 async function step(n) {{
    if (n == -1) {{
@@ -151,14 +156,14 @@ async function step(n) {{
    let id = steps[n];
    return await Deno.core.opAsync("op_get_result", [workspace, id, token, base_url]);
 }}"#,
-            steps.into_iter().map(|x| format!("\"{x}\"")).join(",")
-        )
-    } else {
-        "".to_string()
-    };
+                steps.into_iter().map(|x| format!("\"{x}\"")).join(",")
+            )
+        } else {
+            String::new()
+        };

-    let code = format!(
-        r#"
+        let api_code = format!(
+            r#"
 let workspace = "{workspace}";
 let base_url = "{}";
 async function variable(path) {{
@@ -169,14 +174,24 @@ async function resource(path) {{
    let token = "{token}";
    return await Deno.core.opAsync("op_resource", [workspace, path, token, base_url]);
 }}
+        "#,
+            std::env::var("BASE_INTERNAL_URL")
+                .unwrap_or_else(|_| "http://missing-base-url".to_string()),
+        );
+        (steps_code, api_code)
+    } else {
+        (String::new(), String::new())
+    };
+
+    let code = format!(
+        r#"
+{api_code}
 {}
 {steps_code}
 (async () => {{ 
    {expr} 
 }})()
        "#,
-        std::env::var("BASE_INTERNAL_URL")
-            .unwrap_or_else(|_| "http://missing-base-url".to_string()),
        env.into_iter()
            .map(|(a, b)| format!(
                "let {a} = {};\n",
@@ -259,7 +274,7 @@ mod tests {
        let code = "value.test + params.test";

        let mut runtime = JsRuntime::new(RuntimeOptions::default());
-        let res = eval(&mut runtime, code, env, "workspace", "token", vec![]).await?;
+        let res = eval(&mut runtime, code, env, None, vec![]).await?;
        assert_eq!(res, json!(4));
        Ok(())
    }
@@ -272,7 +287,7 @@ mod tests {
 multiline template`";

        let mut runtime = JsRuntime::new(RuntimeOptions::default());
-        let res = eval(&mut runtime, code, env, "workspace", "token", vec![]).await?;
+        let res = eval(&mut runtime, code, env, None, vec![]).await?;
        assert_eq!(res, json!("my 5\nmultiline template"));
        Ok(())
    }
@@ -285,7 +300,7 @@ multiline template`";
        ];
        let code = r#"variable("test")"#;

-        let res = eval_timeout(code.to_string(), env, "workspace", "token", vec![]).await?;
+        let res = eval_timeout(code.to_string(), env, None, vec![]).await?;
        assert_eq!(res, json!("test"));
        Ok(())
    }
--- a/backend/src/lib.rs
+++ b/backend/src/lib.rs
@@ -9,15 +9,13 @@ use argon2::Argon2;
 use axum::{handler::Handler, middleware::from_extractor, routing::get, Extension, Router};
 use db::DB;
 use git_version::git_version;
-use hyper::Response;
 use slack_http_verifier::SlackVerifier;
 use std::{net::SocketAddr, sync::Arc};
 use tokio::sync::Mutex;
 use tower::ServiceBuilder;
 use tower_cookies::CookieManagerLayer;
-use tower_http::trace::{MakeSpan, OnResponse, TraceLayer};
-use tracing::{field, Span};
-use tracing_subscriber::{filter::filter_fn, prelude::*, EnvFilter};
+use tower_http::trace::TraceLayer;
+
 extern crate magic_crypt;

 extern crate dotenv;
@@ -38,85 +36,31 @@ mod resources;
 mod schedule;
 mod scripts;
 mod static_assets;
+mod tracing_init;
 mod users;
 mod utils;
 mod variables;
 mod worker;
+mod worker_flow;
 mod worker_ping;
 mod workspaces;

 use error::Error;

 pub use crate::email::EmailSender;
-use crate::{db::UserDB, error::to_anyhow, oauth2::build_oauth_clients, utils::rd_string};
+use crate::{
+    db::UserDB,
+    error::to_anyhow,
+    oauth2::build_oauth_clients,
+    tracing_init::{MyMakeSpan, MyOnResponse},
+    utils::rd_string,
+};

 const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
 pub const DEFAULT_NUM_WORKERS: usize = 3;
 pub const DEFAULT_TIMEOUT: i32 = 300;
 pub const DEFAULT_SLEEP_QUEUE: u64 = 50;

-#[derive(Clone)]
-struct MyOnResponse {}
-
-impl<B> OnResponse<B> for MyOnResponse {
-    fn on_response(
-        self,
-        response: &Response<B>,
-        latency: std::time::Duration,
-        _span: &tracing::Span,
-    ) {
-        tracing::info!(
-            latency = %latency.as_millis(),
-            status = ?response.status(),
-            "finished processed request")
-    }
-}
-
-#[derive(Clone)]
-struct MyMakeSpan {}
-
-impl<B> MakeSpan<B> for MyMakeSpan {
-    fn make_span(&mut self, request: &hyper::Request<B>) -> Span {
-        tracing::info_span!(
-            "request",
-            method = %request.method(),
-            uri = %request.uri(),
-            version = ?request.version(),
-            username = field::Empty,
-        )
-    }
-}
-
-pub async fn initialize_tracing() -> anyhow::Result<()> {
-    //let log_level = if std::env::var("RUST_LOG").map(|x| &x == "debug")
-    let ts_base = tracing_subscriber::registry()
-        .with(
-            EnvFilter::from_default_env()
-                //.add_directive("windmill".parse()?)
-                .add_directive("runtime=trace".parse()?)
-                .add_directive("tokio=trace".parse()?),
-        )
-        .with(
-            tracing_subscriber::fmt::layer()
-                .json()
-                .flatten_event(true)
-                .with_span_list(false)
-                .with_current_span(true)
-                .with_filter(filter_fn(|meta| meta.target().starts_with("windmill"))),
-        );
-
-    if std::env::var("TOKIO_CONSOLE")
-        .map(|x| x == "true")
-        .unwrap_or(false)
-    {
-        let console_layer = console_subscriber::spawn();
-        ts_base.with(console_layer).init();
-    } else {
-        ts_base.init();
-    }
-    Ok(())
-}
-
 pub async fn migrate_db(db: &DB) -> anyhow::Result<()> {
    let app_password = std::env::var("APP_USER_PASSWORD").unwrap_or_else(|_| "changeme".to_owned());

@@ -131,6 +75,10 @@ pub async fn connect_db() -> anyhow::Result<DB> {
    Ok(db::connect(&database_url).await?)
 }

+pub async fn initialize_tracing() -> anyhow::Result<()> {
+    tracing_init::initialize_tracing().await
+}
+
 #[derive(Clone)]
 struct BaseUrl(String);

--- a/backend/src/scripts.rs
+++ b/backend/src/scripts.rs
@@ -128,6 +128,7 @@ pub struct Script {
    pub lock: Option<String>,
    pub lock_error_logs: Option<String>,
    pub language: ScriptLang,
+    pub trigger_reco_interval: Option<i32>,
 }

 #[derive(Serialize, Deserialize, sqlx::Type, Debug)]
@@ -154,6 +155,7 @@ pub struct NewScript {
    pub is_template: Option<bool>,
    pub lock: Option<Vec<String>>,
    pub language: ScriptLang,
+    pub trigger_reco_interval: Option<i32>,
 }

 #[derive(Deserialize)]
@@ -198,6 +200,7 @@ async fn list_scripts(
            "null as lock",
            "CASE WHEN lock_error_logs IS NOT NULL THEN 'error' ELSE null END as lock_error_logs",
            "language",
+            "trigger_reco_interval",
        ])
        .order_by("created_at", lq.order_desc.unwrap_or(true))
        .and_where("workspace_id = ? OR workspace_id = 'starter'".bind(&w_id))
@@ -404,8 +407,8 @@ async fn create_script(
    //::text::json is to ensure we use serde_json with preserve order
    sqlx::query!(
        "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, \
-         created_by, schema, is_template, extra_perms, lock, language) VALUES \
-         ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13)",
+         created_by, schema, is_template, extra_perms, lock, language, trigger_reco_interval) VALUES \
+         ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)",
        &w_id,
        &hash.0,
        ns.path,
@@ -418,7 +421,8 @@ async fn create_script(
        ns.is_template.unwrap_or(false),
        extra_perms,
        lock,
-        ns.language: ScriptLang
+        ns.language: ScriptLang,
+        ns.trigger_reco_interval,
    )
    .execute(&mut tx)
    .await?;
--- a/backend/src/tracing_init.rs
+++ b/backend/src/tracing_init.rs
@@ -0,0 +1,95 @@
+use ::tracing::{field, Metadata, Span};
+use ::tracing_subscriber::{
+    filter::filter_fn,
+    fmt::{format, Layer},
+    prelude::*,
+    EnvFilter,
+};
+use hyper::Response;
+use tower_http::trace::{MakeSpan, OnResponse};
+
+#[derive(Clone)]
+pub struct MyOnResponse {}
+
+impl<B> OnResponse<B> for MyOnResponse {
+    fn on_response(
+        self,
+        response: &Response<B>,
+        latency: std::time::Duration,
+        _span: &tracing::Span,
+    ) {
+        tracing::info!(
+            latency = %latency.as_millis(),
+            status = ?response.status(),
+            "finished processed request")
+    }
+}
+
+#[derive(Clone)]
+pub struct MyMakeSpan {}
+
+impl<B> MakeSpan<B> for MyMakeSpan {
+    fn make_span(&mut self, request: &hyper::Request<B>) -> Span {
+        tracing::info_span!(
+            "request",
+            method = %request.method(),
+            uri = %request.uri(),
+            version = ?request.version(),
+            username = field::Empty,
+        )
+    }
+}
+
+fn json_layer<S>() -> Layer<S, format::JsonFields, format::Format<format::Json>> {
+    tracing_subscriber::fmt::layer()
+        .json()
+        .flatten_event(true)
+        .with_span_list(false)
+        .with_current_span(true)
+}
+
+fn compact_layer<S>() -> Layer<S, format::DefaultFields, format::Format<format::Compact>> {
+    tracing_subscriber::fmt::layer().compact()
+}
+
+fn filter_metadata(meta: &Metadata) -> bool {
+    meta.target().starts_with("windmill")
+}
+
+pub async fn initialize_tracing() -> anyhow::Result<()> {
+    let tokio_console = std::env::var("TOKIO_CONSOLE")
+        .map(|x| x == "true")
+        .unwrap_or(false);
+    let json_fmt = std::env::var("JSON_FMT")
+        .map(|x| x == "true")
+        .unwrap_or(false);
+
+    let env_filter = EnvFilter::from_default_env();
+
+    let nenv_filter = if tokio_console {
+        env_filter
+            .add_directive("runtime=trace".parse()?)
+            .add_directive("tokio=trace".parse()?)
+    } else {
+        env_filter
+    };
+    let ts_base = tracing_subscriber::registry().with(nenv_filter);
+
+    match (json_fmt, tokio_console) {
+        (true, true) => ts_base
+            .with(json_layer().with_filter(filter_fn(filter_metadata)))
+            .with(console_subscriber::spawn())
+            .init(),
+        (true, false) => ts_base
+            .with(json_layer().with_filter(filter_fn(filter_metadata)))
+            .init(),
+        (false, true) => ts_base
+            .with(compact_layer().with_filter(filter_fn(filter_metadata)))
+            .with(console_subscriber::spawn())
+            .init(),
+        _ => ts_base
+            .with(compact_layer().with_filter(filter_fn(filter_metadata)))
+            .init(),
+    }
+    Ok(())
+}
--- a/backend/src/variables.rs
+++ b/backend/src/variables.rs
@@ -86,7 +86,8 @@ pub fn get_reserved_variables(
    permissioned_as: &str,
    path: Option<String>,
    flow_path: Option<String>,
-) -> [ContextualVariable; 8] {
+    schedule_path: Option<String>,
+) -> [ContextualVariable; 9] {
    [
        ContextualVariable {
            name: "WM_WORKSPACE".to_string(),
@@ -123,6 +124,11 @@ pub fn get_reserved_variables(
            value: flow_path.unwrap_or_else(|| "".to_string()),
            description: "Path of the encapsulating flow if the job is a flow step".to_string()
        },
+        ContextualVariable {
+            name: "WM_SCHEDULE_PATH".to_string(),
+            value: schedule_path.unwrap_or_else(|| "".to_string()),
+            description: "Path of the schedule if the job of the step or encapsulating step has been triggered by a schedule".to_string()
+        },
        ContextualVariable {
            name: "WM_PERMISSIONED_AS".to_string(),
            value: permissioned_as.to_string(),
@@ -147,6 +153,7 @@ async fn list_contextual_variables(
            format!("u/{username}").as_str(),
            Some("u/user/script_path".to_string()),
            Some("u/user/encapsulating_flow_path".to_string()),
+            Some("u/user/triggering_flow_path".to_string()),
        )
        .to_vec(),
    ))
--- a/backend/src/worker.rs
+++ b/backend/src/worker.rs
@@ -20,14 +20,16 @@ use crate::{
    db::DB,
    error::Error,
    jobs::{
-        add_completed_job, add_completed_job_error, handle_flow, postprocess_queued_job, pull,
-        update_flow_status_after_job_completion, update_flow_status_in_progress, JobKind,
+        add_completed_job, add_completed_job_error, postprocess_queued_job, pull, JobKind,
        QueuedJob,
    },
    parser::{self, Typ},
    scripts::{ScriptHash, ScriptLang},
    users::{create_token_for_owner, get_email_from_username},
    variables,
+    worker_flow::{
+        handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress,
+    },
 };

 use serde_json::{json, Map, Value};
@@ -131,9 +133,14 @@ pub async fn run_worker(
                    )
                    .await;

-                    let _ =
-                        postprocess_queued_job(job2.schedule_path, &job2.workspace_id, job2.id, db)
-                            .await;
+                    let _ = postprocess_queued_job(
+                        job2.schedule_path,
+                        job2.script_path,
+                        &job2.workspace_id,
+                        job2.id,
+                        db,
+                    )
+                    .await;
                    tracing::error!(job_id = %job2.id, "Error handling job: {err_string}");
                };
            }
@@ -217,7 +224,7 @@ async fn handle_queued_job(

            match execution {
                Ok(r) => {
-                    add_completed_job(db, &job, true, r.result.clone(), logs).await?;
+                    add_completed_job(db, &job, true, false, r.result.clone(), logs).await?;
                    if job.is_flow_step {
                        update_flow_status_after_job_completion(db, &job, true, r.result).await?;
                    }
@@ -231,7 +238,8 @@ async fn handle_queued_job(
                }
            };

-            let _ = postprocess_queued_job(job.schedule_path, &w_id, job_id, db).await;
+            let _ =
+                postprocess_queued_job(job.schedule_path, job.script_path, &w_id, job_id, db).await;
        }
    }
    Ok(())
@@ -485,8 +493,8 @@ async fn handle_nondep_job(
                } else {
                    None
                };
-                let ser_args = serde_json::to_string(&args)
-                    .map_err(|e| Error::ExecutionErr(e.to_string()))?;
+                let ser_args =
+                    serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?;
                write_file(job_dir, "args.json", &ser_args).await?;

                let wrapper_content: String = format!(
@@ -596,8 +604,8 @@ print(res_json)
            } else {
                None
            };
-            let ser_args = serde_json::to_string(&args)
-                .map_err(|e| Error::ExecutionErr(e.to_string()))?;
+            let ser_args =
+                serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?;
            write_file(job_dir, "args.json", &ser_args).await?;

            let spread = sig.args.into_iter().map(|x| x.name).join(",");
@@ -630,7 +638,9 @@ run();
            write_file(job_dir, "main.ts", &wrapper_content).await?;

            tx.commit().await?;
-            let reserved_variables = get_reserved_variables(job, token, db).await?;
+
+            let mut reserved_variables = get_reserved_variables(job, token, db).await?;
+            reserved_variables.insert("RUST_LOG".to_string(), "info".to_string());

            if !disable_nuser {
                let _ = write_file(
@@ -767,6 +777,7 @@ async fn get_reserved_variables(
        &job.permissioned_as,
        job.script_path.clone(),
        flow_path,
+        job.schedule_path.clone(),
    );
    Ok(variables
        .into_iter()
--- a/backend/src/worker_flow.rs
+++ b/backend/src/worker_flow.rs
@@ -0,0 +1,606 @@
+use std::collections::HashMap;
+
+use crate::flow::{FlowModuleValue, FlowValue, InputTransform};
+use crate::jobs::{
+    add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push,
+    script_path_to_payload, JobPayload,
+};
+use crate::js_eval::{eval_timeout, EvalCreds};
+use crate::users::create_token_for_owner;
+use crate::{
+    db::DB,
+    error::{self, Error},
+    jobs::QueuedJob,
+};
+use async_recursion::async_recursion;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use serde_json::{Map, Value};
+use tracing::instrument;
+use uuid::Uuid;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct FlowStatus {
+    pub step: i32,
+    pub modules: Vec<FlowStatusModule>,
+    pub failure_module: FlowStatusModule,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Iterator {
+    pub index: u8,
+    pub itered: Vec<Value>,
+    pub args: Map<String, Value>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(tag = "type")]
+pub enum FlowStatusModule {
+    WaitingForPriorSteps,
+    WaitingForEvent {
+        event: String,
+    },
+    WaitingForExecutor {
+        job: Uuid,
+    },
+    InProgress {
+        job: Uuid,
+        iterator: Option<Iterator>,
+        forloop_jobs: Option<Vec<Uuid>>,
+    },
+    Success {
+        job: Uuid,
+        forloop_jobs: Option<Vec<Uuid>>,
+    },
+    Failure {
+        job: Uuid,
+        forloop_jobs: Option<Vec<Uuid>>,
+    },
+}
+
+#[async_recursion]
+#[instrument(level = "trace", skip_all)]
+pub async fn update_flow_status_after_job_completion(
+    db: &DB,
+    job: &QueuedJob,
+    success: bool,
+    result: Option<Map<String, Value>>,
+) -> error::Result<()> {
+    tracing::info!("HANDLE FLOW: {job:?} {success} {result:?}");
+
+    let mut tx = db.begin().await?;
+
+    let w_id = &job.workspace_id;
+
+    let flow = job
+        .parent_job
+        .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?;
+
+    let old_status_json = sqlx::query_scalar!(
+        "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2",
+        flow,
+        w_id
+    )
+    .fetch_one(&mut tx)
+    .await?
+    .ok_or_else(|| Error::InternalErr(format!("requiring a previous status")))?;
+
+    let old_status = serde_json::from_value::<FlowStatus>(old_status_json)
+        .ok()
+        .ok_or_else(|| {
+            Error::InternalErr(format!("requiring status to be parsabled as FlowStatus"))
+        })?;
+
+    let (step_counter, new_status) = match &old_status.modules[old_status.step as usize] {
+        module_status @ FlowStatusModule::InProgress {
+            iterator: Some(Iterator { index, itered, .. }),
+            ..
+        } if (index.to_owned() as usize) < itered.len() - 1 && success => {
+            (old_status.step, module_status.clone())
+        }
+        module_status @ _ => {
+            let forloop_jobs = match module_status {
+                FlowStatusModule::InProgress {
+                    forloop_jobs: Some(jobs),
+                    ..
+                } => Some(jobs.clone()),
+                _ => None,
+            };
+            let new_status = if success {
+                FlowStatusModule::Success {
+                    job: job.id,
+                    forloop_jobs,
+                }
+            } else {
+                FlowStatusModule::Failure {
+                    job: job.id,
+                    forloop_jobs,
+                }
+            };
+            (old_status.step + 1, new_status)
+        }
+    };
+
+    let last_step = step_counter as usize == old_status.modules.len();
+
+    tracing::debug!(
+        "old status: {:#?}\n{:#?}\n{last_step}",
+        old_status,
+        new_status
+    );
+
+    let prev_step = old_status.step;
+    let (stop_early_expr, skip_if_stop_early) = sqlx::query_as::<_, (Option<String>, Option<bool>)>(&format!(
+        "UPDATE queue
+            SET 
+                flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {prev_step}}}', $1), '{{\"step\"}}', $2)
+            WHERE id = $3
+            RETURNING 
+                (raw_flow->'modules'->{prev_step}->>'stop_after_if_expr'), 
+                (raw_flow->'modules'->{prev_step}->>'skip_if_stopped')::bool",
+    ))
+    .bind(serde_json::json!(new_status))
+    .bind(serde_json::json!(step_counter))
+    .bind(flow)
+    .fetch_one(&mut tx)
+    .await?;
+
+    tracing::info!("UPDATE: {:?}", new_status);
+
+    let flow_job = get_queued_job(flow, w_id, &mut tx)
+        .await?
+        .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
+    tx.commit().await?;
+
+    let stop_early = success
+        && if let Some(expr) = stop_early_expr {
+            compute_stop_early(expr, result.clone()).await?
+        } else {
+            false
+        };
+
+    let done = if !success || last_step || stop_early {
+        let result = match new_status {
+            FlowStatusModule::Success {
+                forloop_jobs: Some(jobs),
+                ..
+            } => {
+                let mut results = Vec::new();
+                for job in jobs {
+                    let result = sqlx::query_scalar!(
+                        "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
+                        job,
+                        w_id,
+                    )
+                    .fetch_optional(db)
+                    .await?
+                    .flatten();
+                    results.push(result.clone());
+                }
+                let mut results_map = serde_json::Map::new();
+                results_map.insert("res1".to_string(), serde_json::json!(results));
+                Some(results_map)
+            }
+            _ => result.clone(),
+        };
+
+        let logs = if stop_early {
+            "Flow job stopped early".to_string()
+        } else {
+            "Flow job completed".to_string()
+        };
+        tracing::debug!("{skip_if_stop_early:?}");
+        add_completed_job(
+            db,
+            &flow_job,
+            success,
+            stop_early && skip_if_stop_early.unwrap_or(false),
+            result.clone(),
+            logs,
+        )
+        .await?;
+        true
+    } else {
+        match handle_flow(&flow_job, db, result.clone()).await {
+            Err(err) => {
+                let _ = add_completed_job_error(
+                    db,
+                    &flow_job,
+                    "Unexpected error during flow chaining:\n".to_string(),
+                    err,
+                )
+                .await;
+                true
+            }
+            Ok(_) => false,
+        }
+    };
+
+    if done {
+        postprocess_queued_job(
+            flow_job.schedule_path.clone(),
+            flow_job.script_path.clone(),
+            &w_id,
+            flow,
+            db,
+        )
+        .await?;
+
+        if flow_job.parent_job.is_some() {
+            return Ok(
+                update_flow_status_after_job_completion(db, &flow_job, success, result).await?,
+            );
+        }
+    }
+
+    Ok(())
+}
+
+async fn compute_stop_early(
+    expr: String,
+    result: Option<Map<String, Value>>,
+) -> error::Result<bool> {
+    let result = serde_json::Value::Object(result.clone().unwrap_or_else(|| Map::new()));
+    match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? {
+        serde_json::Value::Bool(true) => Ok(true),
+        serde_json::Value::Bool(false) => Ok(false),
+        a @ _ => Err(Error::ExecutionErr(format!(
+            "Expected a boolean value, found: {a:?}"
+        ))),
+    }
+}
+
+pub fn init_flow_status(f: &FlowValue) -> FlowStatus {
+    FlowStatus {
+        step: 0,
+        modules: (0..f.modules.len())
+            .map(|_| FlowStatusModule::WaitingForPriorSteps)
+            .collect(),
+        failure_module: FlowStatusModule::WaitingForPriorSteps,
+    }
+}
+
+pub async fn update_flow_status_in_progress(
+    db: &DB,
+    w_id: &str,
+    flow: Uuid,
+    job_in_progress: Uuid,
+) -> error::Result<()> {
+    let step = get_step_of_flow_status(db, flow).await?;
+    sqlx::query(&format!(
+        "UPDATE queue
+            SET flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1)
+            WHERE id = $2 AND workspace_id = $3",
+        step
+    ))
+    .bind(serde_json::json!(FlowStatusModule::InProgress {
+        job: job_in_progress,
+        iterator: None,
+        forloop_jobs: None,
+    }))
+    .bind(flow)
+    .bind(w_id)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+#[instrument(level = "trace", skip_all)]
+pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<i32> {
+    let r = sqlx::query_scalar!(
+        "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1",
+        id
+    )
+    .fetch_one(db)
+    .await?
+    .ok_or_else(|| Error::InternalErr(format!("not found step")))?;
+    Ok(r)
+}
+
+#[instrument(level = "trace", skip_all)]
+async fn transform_input(
+    flow_args: &Option<serde_json::Value>,
+    last_result: Option<Map<String, serde_json::Value>>,
+    input_transform: &HashMap<String, InputTransform>,
+    workspace: &str,
+    token: &str,
+    steps: Vec<String>,
+) -> anyhow::Result<Option<Map<String, serde_json::Value>>> {
+    let mut mapped = serde_json::Map::new();
+
+    for (key, val) in input_transform.into_iter() {
+        match val {
+            InputTransform::Static { value } => {
+                mapped.insert(key.to_string(), value.to_owned());
+                ()
+            }
+            _ => (),
+        };
+    }
+
+    for (key, val) in input_transform.into_iter() {
+        match val {
+            InputTransform::Static { value: _ } => (),
+            InputTransform::Javascript { expr } => {
+                let previous_result =
+                    serde_json::Value::Object(last_result.clone().unwrap_or_else(|| Map::new()));
+                let flow_input = flow_args.clone().unwrap_or_else(|| json!({}));
+                let v = eval_timeout(
+                    expr.to_string(),
+                    vec![
+                        ("params".to_string(), serde_json::json!(mapped)),
+                        ("previous_result".to_string(), previous_result),
+                        ("flow_input".to_string(), flow_input),
+                    ],
+                    Some(EvalCreds {
+                        workspace: workspace.to_string(),
+                        token: token.to_string(),
+                    }),
+                    steps.clone(),
+                )
+                .await
+                .map_err(|e| {
+                    Error::ExecutionErr(format!(
+                        "Error during isolated evaluation of expression `{expr}`:\n{e}"
+                    ))
+                })?;
+                mapped.insert(key.to_string(), v);
+                ()
+            }
+        }
+    }
+
+    Ok(Some(mapped))
+}
+
+#[instrument(level = "trace", skip_all)]
+pub async fn handle_flow(
+    flow_job: &QueuedJob,
+    db: &sqlx::Pool<sqlx::Postgres>,
+    last_result: Option<Map<String, serde_json::Value>>,
+) -> anyhow::Result<()> {
+    let value = flow_job
+        .raw_flow
+        .as_ref()
+        .ok_or_else(|| Error::InternalErr(format!("requiring a raw flow value")))?
+        .to_owned();
+    let flow = serde_json::from_value::<FlowValue>(value.to_owned())?;
+    push_next_flow_job(
+        flow_job,
+        flow,
+        flow_job.schedule_path.clone(),
+        db,
+        last_result,
+    )
+    .await?;
+    Ok(())
+}
+
+#[instrument(level = "trace", skip_all)]
+async fn push_next_flow_job(
+    flow_job: &QueuedJob,
+    flow: FlowValue,
+    schedule_path: Option<String>,
+    db: &sqlx::Pool<sqlx::Postgres>,
+    last_result: Option<Map<String, serde_json::Value>>,
+) -> anyhow::Result<()> {
+    let flow_status_json = flow_job.flow_status.as_ref().ok_or_else(|| {
+        Error::InternalErr(format!("not found status for flow job {:?}", flow_job.id))
+    })?;
+    let status = serde_json::from_value::<FlowStatus>(flow_status_json.to_owned())?;
+    let i = status.step as usize;
+
+    if flow.modules.len() > i {
+        let module = &flow.modules[i];
+        let mut tx = db.begin().await?;
+        let job_payload = match &module.value {
+            FlowModuleValue::Script {
+                path: script_path
+            } => script_path_to_payload(script_path, &mut tx, &flow_job.workspace_id).await?,
+            FlowModuleValue::RawScript(raw_code) => {
+                let mut raw_code = raw_code.clone();
+                if raw_code.path.is_none() {
+                    raw_code.path = Some(format!(
+                        "{}/{i}",
+                        flow_job
+                            .script_path
+                            .as_ref()
+                            .unwrap_or(&"NO_FLOW_PATH".to_owned())
+                    ));
+                }
+                JobPayload::Code(raw_code)
+            }
+            FlowModuleValue::ForloopFlow { iterator: _, value } => JobPayload::RawFlow {
+                value: *(*value).to_owned(),
+                path: Some(format!(
+                    "{}/{i}",
+                    flow_job
+                        .script_path
+                        .as_ref()
+                        .unwrap_or(&"NO_FLOW_PATH".to_owned())
+                )),
+            },
+            a @ _ => {
+                tracing::info!("Unrecognized module values {:?}", a);
+                Err(Error::BadRequest(format!(
+                    "Unrecognized module values {:?}",
+                    a
+                )))?
+            }
+        };
+
+        let token = create_token_for_owner(
+            &db,
+            &flow_job.workspace_id,
+            &flow_job.permissioned_as,
+            crate::users::NewToken {
+                label: Some("transform-input".to_string()),
+                expiration: Some(chrono::Utc::now() + chrono::Duration::seconds(10)),
+            },
+            &flow_job.created_by,
+        )
+        .await?;
+        let mut input_transform = module.input_transform.clone();
+
+        tracing::debug!(
+            "PUSH: module: {:#?}, status: {:#?}",
+            module.value,
+            status.modules[i]
+        );
+        let (forloop_args, forloop_iterator) = match &module.value {
+            FlowModuleValue::ForloopFlow { iterator, .. } => {
+                let (index_forloop, itered, args, forloop_jobs) = match &status.modules[i] {
+                    FlowStatusModule::WaitingForPriorSteps { .. } => {
+                        let itered = match iterator {
+                            InputTransform::Static { value } => value.clone(),
+                            InputTransform::Javascript { expr } => {
+                                let result = serde_json::Value::Object(
+                                    last_result.clone().unwrap_or_else(|| Map::new()),
+                                );
+                                eval_timeout(
+                                    expr.to_string(),
+                                    [("result".to_string(), result)].into(),
+                                    None,
+                                    vec![],
+                                )
+                                .await?
+                            }
+                        };
+                        input_transform.insert(
+                            "_index".to_string(),
+                            InputTransform::Static {
+                                value: serde_json::Value::Number(serde_json::Number::from(0)),
+                            },
+                        );
+                        input_transform.insert(
+                            "_value".to_string(),
+                            InputTransform::Static {
+                                value: itered[0].clone(),
+                            },
+                        );
+
+                        match itered {
+                            serde_json::Value::Array(arr) => (0 as u8, arr, None, vec![]),
+                            a @ _ => Err(Error::BadRequest(format!(
+                                "Expected an array value, found: {:?}",
+                                a
+                            )))?,
+                        }
+                    }
+                    FlowStatusModule::InProgress {
+                        iterator:
+                            Some(Iterator {
+                                index,
+                                itered,
+                                args,
+                            }),
+                        forloop_jobs,
+                        ..
+                    } => {
+                        let mut args = args.clone();
+                        let nindex = index.to_owned() + 1;
+                        args.insert(
+                            "_index".to_string(),
+                            serde_json::Value::Number(serde_json::Number::from(nindex.to_owned())),
+                        );
+                        args.insert(
+                            "_value".to_string(),
+                            itered[nindex.to_owned() as usize].clone(),
+                        );
+                        (
+                            nindex,
+                            itered.to_owned(),
+                            Some(args),
+                            forloop_jobs.to_owned().unwrap_or_else(Vec::new),
+                        )
+                    }
+                    a @ _ => Err(Error::BadRequest(format!(
+                        "Unrecognized module status for ForloopFlow {:?}",
+                        a
+                    )))?,
+                };
+                (args, Some((index_forloop, itered, forloop_jobs)))
+            }
+            _ => (None, None),
+        };
+
+        let args = if forloop_args.is_some() {
+            forloop_args.map(|x| x.to_owned())
+        } else {
+            let steps = status
+                .modules
+                .into_iter()
+                .map(|x| match x {
+                    FlowStatusModule::Success {
+                        job,
+                        forloop_jobs: _,
+                    } => job.to_string(),
+                    _ => "invalid step status".to_string(),
+                })
+                .collect();
+
+            let transformed = transform_input(
+                &flow_job.args,
+                last_result.clone(),
+                &input_transform,
+                &flow_job.workspace_id,
+                &token,
+                steps,
+            )
+            .await?;
+
+            match (&flow_job.args, &module.value) {
+                (Some(Value::Object(m)), FlowModuleValue::ForloopFlow { .. }) => {
+                    let mut args = transformed.unwrap_or_else(Map::new);
+                    args.extend(m.to_owned());
+                    args.extend(last_result.unwrap());
+                    Some(args)
+                }
+                _ => transformed,
+            }
+        };
+
+        let (uuid, mut tx) = push(
+            tx,
+            &flow_job.workspace_id,
+            job_payload,
+            args.clone(),
+            &flow_job.created_by,
+            flow_job.permissioned_as.to_owned(),
+            None,
+            schedule_path,
+            Some(flow_job.id),
+            true,
+        )
+        .await?;
+
+        let new_status = if let Some((index, itered, mut forloop_jobs)) = forloop_iterator {
+            forloop_jobs.push(uuid.to_owned());
+            serde_json::json!(FlowStatusModule::InProgress {
+                job: uuid,
+                iterator: Some(Iterator {
+                    index: index,
+                    itered: itered,
+                    args: args.unwrap_or_else(|| Map::new()),
+                }),
+                forloop_jobs: Some(forloop_jobs),
+            })
+        } else {
+            serde_json::json!(FlowStatusModule::WaitingForExecutor { job: uuid })
+        };
+
+        sqlx::query(&format!(
+            "UPDATE queue
+            SET 
+                flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1)
+            WHERE id = $2",
+            i
+        ))
+        .bind(new_status)
+        .bind(flow_job.id)
+        .execute(&mut tx)
+        .await?;
+        tx.commit().await?;
+    }
+    Ok(())
+}
--- a/deno-client/generate.sh
+++ b/deno-client/generate.sh
@@ -3,4 +3,4 @@ set -e

 /usr/local/bin/docker-entrypoint.sh generate -i ../backend/openapi.yaml -g typescript --additional-properties platform=deno -o windmill-api
 sed -i 's/this\.type = "Job";//' windmill-api/models/Job.ts
-sed -i '457 i \        if (mediaType === "text/plain") { return rawData }' windmill-api/models/ObjectSerializer.ts
+sed -i '460 i \        if (mediaType === "text/plain") { return data }' windmill-api/models/ObjectSerializer.ts
--- a/deno-client/mod.ts
+++ b/deno-client/mod.ts
@@ -52,11 +52,13 @@ export function getInternalStatePath(suffix?: string): string {
    const permissioned_as = Deno.env.get("WM_PERMISSIONED_AS")
    const flow_path = env_flow_path != undefined && env_flow_path != "" ? env_flow_path : 'NO_FLOW_PATH'
    const script_path = suffix ?? (env_job_path != undefined && env_job_path != "" ? env_job_path : 'NO_JOB_PATH')
+    const env_schedule_path = Deno.env.get("WM_SCHEDULE_PATH")
+    const schedule_path = env_flow_path != undefined && env_flow_path != "" ? `/${env_schedule_path}` : ''

    if (script_path.slice(script_path.length - 1) === '/') {
        throw Error(`The script path must not end with '/', give a name to your script!`)
    }
-    return `${permissioned_as}/${flow_path}/${script_path}`
+    return `${permissioned_as}/${flow_path}/${script_path}${schedule_path}`
 }

 /**
--- a/frontend/src/lib/components/DisplayResult.svelte
+++ b/frontend/src/lib/components/DisplayResult.svelte
@@ -57,52 +57,58 @@
 <svelte:head>
 	{@html github}
 </svelte:head>
-{#if Object.keys(result).length > 0}<div>
-		The result keys are: <b>{Object.keys(result).join(', ')}</b>
-	</div>
-{/if}
-{#if resultKind == 'table-col'}
-	<div class="grid grid-flow-col-dense border border-gray-200 rounded-md ">
-		{#each Object.keys(result) as col}
-			<div class="flex flex-col  min-w-full">
-				<div class="px-12 text-left uppercase border-b bg-gray-50 overflow-hidden rounded-t-md ">
-					{col}
-				</div>
-				{#each result[col] as item}
-					<div class="px-12 text-left">{item}</div>
-				{/each}
-			</div>
-		{/each}
-	</div>
-{:else if resultKind == 'table-row'}<div
-		class="grid grid-flow-col-dense border border-gray-200 rounded-md "
-	>
-		<TableCustom>
-			<tbody slot="body">
-				{#each asListOfList(Object.values(result)[0]) as row}
-					<tr>
-						{#each row as v}
-							<td>{v ?? ''}</td>
+{#if result}
+	{#if Object.keys(result).length > 0}<div>
+			The result keys are: <b>{Object.keys(result).join(', ')}</b>
+		</div>
+	{/if}
+	{#if resultKind == 'table-col'}
+		<div class="grid grid-flow-col-dense border border-gray-200 rounded-md ">
+			{#each Object.keys(result) as col}
+				<div class="flex flex-col  min-w-full">
+					<div class="px-12 text-left uppercase border-b bg-gray-50 overflow-hidden rounded-t-md ">
+						{col}
+					</div>
+					{#if Array.isArray(result[col])}
+						{#each result[col] as item}
+							<div class="px-12 text-left">
+								{typeof item === 'string' ? item : JSON.stringify(item)}
+							</div>
 						{/each}
-					</tr>
-				{/each}
-			</tbody>
-		</TableCustom>
-	</div>
-{:else if resultKind == 'png'}
-	<div class="h-full">
-		Result is an image: <img
-			alt="png rendered"
-			class="w-auto h-full"
-			src="data:image/png;base64,{result.png}"
-		/>
-	</div>
-{:else if resultKind == 'file'}
-	<div>
-		Result is a file: <a
-			download="windmill.file"
-			href="data:application/octet-stream;base64,{result.file}">Download</a
+					{/if}
+				</div>
+			{/each}
+		</div>
+	{:else if resultKind == 'table-row'}<div
+			class="grid grid-flow-col-dense border border-gray-200 rounded-md "
 		>
-	</div>
-{:else}<Highlight language={json} code={JSON.stringify(result, null, 4).replace(/\\n/g, '\n')} />
+			<TableCustom>
+				<tbody slot="body">
+					{#each asListOfList(Object.values(result)[0]) as row}
+						<tr>
+							{#each row as v}
+								<td>{v ?? ''}</td>
+							{/each}
+						</tr>
+					{/each}
+				</tbody>
+			</TableCustom>
+		</div>
+	{:else if resultKind == 'png'}
+		<div class="h-full">
+			Result is an image: <img
+				alt="png rendered"
+				class="w-auto h-full"
+				src="data:image/png;base64,{result.png}"
+			/>
+		</div>
+	{:else if resultKind == 'file'}
+		<div>
+			Result is a file: <a
+				download="windmill.file"
+				href="data:application/octet-stream;base64,{result.file}">Download</a
+			>
+		</div>
+	{:else}<Highlight language={json} code={JSON.stringify(result, null, 4).replace(/\\n/g, '\n')} />
+	{/if}
 {/if}
--- a/frontend/src/lib/components/FlowBuilder.svelte
+++ b/frontend/src/lib/components/FlowBuilder.svelte
@@ -7,7 +7,8 @@
 	import { onMount } from 'svelte'
 	import SvelteMarkdown from 'svelte-markdown'
 	import FlowEditor from './FlowEditor.svelte'
-	import { flowStore, initFlow } from './flows/flowStore'
+	import { flowStore, initFlow, type FlowMode } from './flows/flowStore'
+	import { flowToMode } from './flows/utils'
 	import Path from './Path.svelte'
 	import Required from './Required.svelte'
 	import ScriptSchema from './ScriptSchema.svelte'
@@ -15,6 +16,8 @@
 	export let flow: Flow
 	export let initialPath: string = ''

+	let mode: FlowMode
+
 	$: step = Number($page.url.searchParams.get('step')) || 1

 	async function loadSearchData() {
@@ -27,27 +30,29 @@
 	}

 	async function saveFlow(): Promise<void> {
+		const newFlow = flowToMode($flowStore, mode)
+
 		if (initialPath === '') {
 			await FlowService.createFlow({
 				workspace: $workspaceStore!,
 				requestBody: {
-					path: flow.path,
-					summary: flow.summary,
-					description: flow.description ?? '',
-					value: flow.value,
-					schema: flow.schema
+					path: newFlow.path,
+					summary: newFlow.summary,
+					description: newFlow.description ?? '',
+					value: newFlow.value,
+					schema: newFlow.schema
 				}
 			})
 		} else {
 			await FlowService.updateFlow({
 				workspace: $workspaceStore!,
-				path: flow.path,
+				path: newFlow.path,
 				requestBody: {
-					path: flow.path,
-					summary: flow.summary,
-					description: flow.description ?? '',
-					value: flow.value,
-					schema: flow.schema
+					path: newFlow.path,
+					summary: newFlow.summary,
+					description: newFlow.description ?? '',
+					value: newFlow.value,
+					schema: newFlow.schema
 				}
 			})
 		}
@@ -60,7 +65,7 @@
 	}

 	flowStore.subscribe((flow: Flow) => {
-		$page.url.searchParams.set('state', btoa(JSON.stringify(flow)))
+		$page.url.searchParams.set('state', btoa(JSON.stringify(flowToMode(flow, mode))))
 		history.replaceState({}, '', $page.url)
 	})

@@ -192,7 +197,7 @@
 			</div>
 		</div>
 	{:else if step === 2}
-		<FlowEditor />
+		<FlowEditor bind:mode />
 	{:else if step === 3}
 		<ScriptSchema
 			synchronizedHeader={false}
--- a/frontend/src/lib/components/FlowEditor.svelte
+++ b/frontend/src/lib/components/FlowEditor.svelte
@@ -1,13 +1,18 @@
 <script lang="ts">
+	import { FlowModuleValue } from '$lib/gen'
+
 	import { faPlus } from '@fortawesome/free-solid-svg-icons'
 	import Icon from 'svelte-awesome'
 	import FlowPreview from './FlowPreview.svelte'
 	import CopyFirstStepSchema from './flows/CopyFirstStepSchema.svelte'
-	import { addModule, flowStore } from './flows/flowStore'
+	import { addModule, flowStore, type FlowMode } from './flows/flowStore'
 	import ModuleStep from './ModuleStep.svelte'
+	import RadioButtonV2 from './RadioButtonV2.svelte'
 	import SchemaEditor from './SchemaEditor.svelte'

 	let args: Record<string, any> = {}
+	export let mode: FlowMode =
+		$flowStore?.value.modules[1]?.value.type == FlowModuleValue.type.FORLOOPFLOW ? 'pull' : 'push'
 	$: numberOfSteps = $flowStore?.value.modules.length - 1
 </script>

@@ -15,6 +20,31 @@
 	<ul class="relative -mt-10">
 		<span class="absolute top-0 left-1/2  h-full w-1 bg-gray-400" aria-hidden="true" />
 		<div class="relative">
+			<li class="flex flex-row flex-shrink max-w-full mx-auto mt-20">
+				<div
+					class="bg-white border border-gray xl-rounded shadow-lg w-full max-w-4xl mx-4 md:mx-auto p4"
+				>
+					<RadioButtonV2
+						options={[
+							[
+								{
+									title: 'Push',
+									desc: 'Trigger this flow through the generated UI, a manual schedule or by calling the associated webhook'
+								},
+								'push'
+							],
+							[
+								{
+									title: 'Pull (WIP)',
+									desc: 'This flow will trigger itself with a schedule to detect changes in external services using a trigger script.'
+								},
+								'pull'
+							]
+						]}
+						bind:value={mode}
+					/>
+				</div>
+			</li>
 			<li class="flex flex-row flex-shrink max-w-full mx-auto mt-20">
 				<div class="bg-white border border-gray xl-rounded shadow-lg w-full mx-4 xl:mx-20">
 					<div
@@ -26,12 +56,12 @@
 					<div class="p-4">
 						<SchemaEditor bind:schema={$flowStore.schema} />
 						<div class="my-4" />
-						<FlowPreview bind:flow={$flowStore} i={numberOfSteps} bind:args />
+						<FlowPreview {mode} bind:flow={$flowStore} i={numberOfSteps} bind:args />
 					</div>
 				</div>
 			</li>
 			{#each $flowStore?.value.modules as mod, i}
-				<ModuleStep bind:mod bind:args {i} />
+				<ModuleStep bind:mod bind:args {i} {mode} />
 			{/each}
 			<li class="relative m-20 ">
 				<div class="relative flex justify-center">
--- a/frontend/src/lib/components/FlowJobResult.svelte
+++ b/frontend/src/lib/components/FlowJobResult.svelte
@@ -0,0 +1,27 @@
+<script lang="ts">
+	import type { CompletedJob } from '$lib/gen'
+
+	import ChevronButton from './ChevronButton.svelte'
+	import DisplayResult from './DisplayResult.svelte'
+
+	export let job: CompletedJob | undefined
+</script>
+
+{#if job}
+	<div class="flex flex-col ml-10">
+		<div>
+			<ChevronButton text="result" viewOptions={true}>
+				<div class="text-xs">
+					<DisplayResult result={job.result} />
+				</div>
+			</ChevronButton>
+		</div>
+		<div>
+			<ChevronButton text="logs" viewOptions={true}>
+				<div class="text-xs p-4 bg-gray-50 overflow-auto max-h-lg">
+					<pre class="w-full">{job.logs}</pre>
+				</div>
+			</ChevronButton>
+		</div>
+	</div>
+{/if}
--- a/frontend/src/lib/components/FlowPreview.svelte
+++ b/frontend/src/lib/components/FlowPreview.svelte
@@ -6,8 +6,9 @@
 	import { faChevronDown, faChevronUp } from '@fortawesome/free-solid-svg-icons'
 	import { createEventDispatcher, onDestroy } from 'svelte'
 	import Icon from 'svelte-awesome'
-	import ChevronButton from './ChevronButton.svelte'
-	import DisplayResult from './DisplayResult.svelte'
+	import FlowJobResult from './FlowJobResult.svelte'
+	import type { FlowMode } from './flows/flowStore'
+	import { flowToMode } from './flows/utils'
 	import FlowStatusViewer from './FlowStatusViewer.svelte'
 	import RunForm from './RunForm.svelte'
 	import Tabs from './Tabs.svelte'
@@ -16,6 +17,7 @@
 	export let i: number
 	export let flow: Flow
 	export let schemas: Schema[] = []
+	export let mode: FlowMode

 	export let args: Record<string, any> = {}

@@ -34,7 +36,8 @@
 	export async function runPreview(args) {
 		viewPreview = true
 		intervalId && clearInterval(intervalId)
-		const newFlow = tab == 'upto' ? truncateFlow(flow) : extractStep(flow)
+		let newFlow = flowToMode(flow, mode)
+		newFlow = tab == 'upto' ? truncateFlow(newFlow) : extractStep(newFlow)
 		jobId = await JobService.runFlowPreview({
 			workspace: $workspaceStore ?? '',
 			requestBody: {
@@ -134,22 +137,7 @@
 			<FlowStatusViewer {job} bind:jobs />
 		</div>
 		{#if `result` in job}
-			<div class="flex flex-col ml-10">
-				<div>
-					<ChevronButton text="result" viewOptions={true}>
-						<div class="text-xs">
-							<DisplayResult result={job.result} />
-						</div>
-					</ChevronButton>
-				</div>
-				<div>
-					<ChevronButton text="logs" viewOptions={true}>
-						<div class="text-xs p-4 bg-gray-50 overflow-auto max-h-lg">
-							<pre class="w-full">{job.logs}</pre>
-						</div>
-					</ChevronButton>
-				</div>
-			</div>
+			<FlowJobResult {job} />
 		{/if}
 	{/if}
 {/if}
--- a/frontend/src/lib/components/FlowStatusViewer.svelte
+++ b/frontend/src/lib/components/FlowStatusViewer.svelte
@@ -7,12 +7,13 @@

 	import { CompletedJob, FlowModuleValue, FlowStatusModule, JobService, QueuedJob } from '$lib/gen'
 	import { workspaceStore } from '$lib/stores'
-	import DisplayResult from './DisplayResult.svelte'
-	import ChevronButton from './ChevronButton.svelte'
 	import JobStatus from './JobStatus.svelte'
+	import FlowJobResult from './FlowJobResult.svelte'

 	export let job: QueuedJob | CompletedJob
-	export let jobs: (CompletedJob | undefined)[]
+	export let jobs: (CompletedJob | CompletedJob[] | undefined)[] = []
+
+	let forloop_selected = ''

 	function loadResults() {
 		job?.flow_status?.modules?.forEach(async (x, i) => {
@@ -20,12 +21,31 @@
 				(i >= jobs.length && x.type == FlowStatusModule.type.SUCCESS) ||
 				x.type == FlowStatusModule.type.FAILURE
 			) {
-				jobs.push(undefined)
-				jobs[i] = await JobService.getCompletedJob({ workspace: $workspaceStore!, id: x.job! })
+				if (x.forloop_jobs) {
+					const forloop_jobs: CompletedJob[] = []
+
+					for (let j of x.forloop_jobs) {
+						forloop_jobs.push(
+							await JobService.getCompletedJob({ workspace: $workspaceStore!, id: j })
+						)
+					}
+					jobs.push(forloop_jobs)
+				} else {
+					jobs.push(await JobService.getCompletedJob({ workspace: $workspaceStore!, id: x.job! }))
+				}
+				jobs = jobs
 			}
 		})
 	}

+	function toCompletedJob(x: any): CompletedJob {
+		return x as CompletedJob
+	}
+
+	function toCompletedJobs(x: any): CompletedJob[] {
+		return x as CompletedJob[]
+	}
+
 	$: $workspaceStore && job && loadResults()
 </script>

@@ -51,7 +71,7 @@
 		out of <span class="font-medium text-gray-900">{job?.raw_flow?.modules.length}</span>
 		<span class="mt-4" />
 	</p>
-	<ul role="list" class="-mb-8 w-full">
+	<ul class="-mb-8 w-full">
 		{#each job?.raw_flow?.modules ?? [] as mod, i}
 			<li class="w-full">
 				<div class="relative pb-8 w-full">
@@ -112,7 +132,15 @@
 							</div>
 							<div class="text-right text-sm whitespace-nowrap text-gray-500">
 								{job.flow_status?.modules[i].type}
-								{#if job.flow_status?.modules[i].job}
+								{#if job.flow_status?.modules[i].forloop_jobs}
+									{#each job.flow_status?.modules[i].forloop_jobs ?? [] as job}
+										<div class="flex flex-col">
+											<a href="/run/{job}" class="font-medium text-blue-600"
+												>{truncateRev(job ?? '', 10)}</a
+											>
+										</div>
+									{/each}
+								{:else if job.flow_status?.modules[i].job}
 									<a href="/run/{job.flow_status?.modules[i].job}" class="font-medium text-blue-600"
 										>{truncateRev(job.flow_status?.modules[i].job ?? '', 10)}</a
 									>
@@ -120,23 +148,29 @@
 							</div>
 						</div>
 					</div>
-					{#if i < jobs.length && jobs[i] != undefined}
-						<div class="flex flex-col ml-10">
-							<div>
-								<ChevronButton text="result" viewOptions={true}>
-									<div class="text-xs">
-										<DisplayResult result={jobs[i]?.result} />
-									</div>
-								</ChevronButton>
+					{#if jobs[i]}
+						{#if Array.isArray(jobs[i])}
+							<div class="flex flex-col space-y-2">
+								{#each toCompletedJobs(jobs[i]) as job, i}
+									<button
+										class="underline text-blue-600 hover:text-blue-700"
+										on:click={() => {
+											if (forloop_selected == job.id) {
+												forloop_selected = ''
+											} else {
+												forloop_selected = job.id
+											}
+										}}
+										>Iteration: #{i}: {job.id} {forloop_selected == job.id ? '(-)' : '(+)'}</button
+									>
+									{#if forloop_selected == job.id}
+										<svelte:self {job} />
+									{/if}
+								{/each}
 							</div>
-							<div>
-								<ChevronButton text="logs">
-									<div class="text-xs p-4 bg-gray-50 overflow-auto max-h-lg">
-										<pre class="w-full">{jobs[i]?.logs}</pre>
-									</div>
-								</ChevronButton>
-							</div>
-						</div>
+						{:else}
+							<FlowJobResult job={toCompletedJob(jobs[i])} />
+						{/if}
 					{/if}
 				</div>
 			</li>
--- a/frontend/src/lib/components/JobStatus.svelte
+++ b/frontend/src/lib/components/JobStatus.svelte
@@ -20,7 +20,7 @@

 {#if job && 'success' in job && job.success}
 	<Icon class="text-green-600" data={check} scale={SMALL_ICON_SCALE} />
-	<span class="mx-2">Succeeded</span>
+	<span class="mx-2">Succeeded {job.is_skipped ? '(Skipped)' : ''}</span>
 	<div>
 		<Icon class="text-gray-700" data={faHourglassHalf} scale={SMALL_ICON_SCALE} /><span class="mx-2"
 			>Job ran in {job.duration}
--- a/frontend/src/lib/components/ModuleStep.svelte
+++ b/frontend/src/lib/components/ModuleStep.svelte
@@ -11,10 +11,12 @@
 		flowStore,
 		loadSchema,
 		pickScript,
-		schemasStore
+		schemasStore,
+		type FlowMode
 	} from './flows/flowStore'
 	import SchemaForm from './SchemaForm.svelte'

+	export let mode: FlowMode
 	export let i: number
 	export let mod: FlowModule
 	export let args: Record<string, any> = {}
@@ -39,8 +41,9 @@
 		<div class="p-6">
 			{#if shouldPick}
 				<FlowInputs
+					isTrigger={mode == 'pull' && i == 0}
 					on:pick={(e) => pickScript(e.detail.path, i)}
-					on:new={(e) => createInlineScriptModule(e.detail.language, i)}
+					on:new={(e) => createInlineScriptModule(e.detail.language, i, mode)}
 				/>
 			{/if}
 			{#if mod.value.type === FlowModuleValue.type.RAWSCRIPT}
@@ -75,6 +78,7 @@
 					bind:args
 					bind:flow={$flowStore}
 					{i}
+					{mode}
 					bind:schemas={$schemasStore}
 					on:change={(e) => {
 						addPreviewResult(e.detail.result, i + 1)
--- a/frontend/src/lib/components/RadioButtonV2.svelte
+++ b/frontend/src/lib/components/RadioButtonV2.svelte
@@ -0,0 +1,34 @@
+<script lang="ts">
+	export let options: [{ title: string; desc?: string }, string][]
+	export let value: any
+
+	import { createEventDispatcher } from 'svelte'
+
+	$: dispatch('change', value)
+
+	const dispatch = createEventDispatcher()
+</script>
+
+<div class="columns-2 m-4">
+	{#each options as [label, val]}
+		<button
+			type="button"
+			on:click={() => {
+				value = val
+			}}
+			class="default-secondary-button-v2 mb-2 w-full"
+			class:selected={value == val}
+		>
+			<h3>{label.title}</h3>
+			<p>
+				{label.desc}
+			</p>
+		</button>
+	{/each}
+</div>
+
+<style>
+	.selected {
+		@apply bg-blue-500/90 text-white;
+	}
+</style>
--- a/frontend/src/lib/components/SchemaForm.svelte
+++ b/frontend/src/lib/components/SchemaForm.svelte
@@ -13,11 +13,11 @@
 	export let schema: Schema
 	export let args: Record<string, InputTransform> = {}
 	export let editableSchema = false
-	export let extraLib: string
+	export let extraLib: string = 'missing extraLib'
 	export let isValid: boolean = true

-	export let i: number
-	export let previousSchema: Object
+	export let i: number | undefined = undefined
+	export let previousSchema: Object | undefined = undefined

 	let inputCheck: { [id: string]: boolean } = {}
 	let editor: Editor
@@ -53,7 +53,7 @@ previous_result.${key}`
 						small={true}
 						bind:value={args[argName].type}
 						on:change={(e) => {
-							args[argName].expr = e.detail == 'javascript' ? getDefaultExpr(i) : undefined
+							args[argName].expr = e.detail == 'javascript' ? getDefaultExpr(i ?? -1) : undefined
 						}}
 					/>
 				</div>
@@ -90,7 +90,7 @@ previous_result.${key}`
 								<PropPicker
 									props={previousSchema}
 									on:select={(event) => {
-										editor.setCode(getDefaultExpr(i, event.detail))
+										editor.setCode(getDefaultExpr(i ?? -1, event.detail))
 									}}
 								/>
 							{:else}
--- a/frontend/src/lib/components/flows/FlowInputs.svelte
+++ b/frontend/src/lib/components/flows/FlowInputs.svelte
@@ -2,25 +2,29 @@
 	import { FlowModuleValue } from '$lib/gen/models/FlowModuleValue'

 	import { faCode } from '@fortawesome/free-solid-svg-icons'
+	import type { integer } from 'monaco-languageclient'
 	import { createEventDispatcher } from 'svelte'
+	import type { FlowMode } from './flowStore'
 	import FlowScriptPicker from './pickers/FlowScriptPicker.svelte'
 	import PickHubScript from './pickers/PickHubScript.svelte'
 	import PickScript from './pickers/PickScript.svelte'

+	export let isTrigger: boolean
+
 	const dispatch = createEventDispatcher()
 </script>

 <div class="columns-2">
-	<PickScript on:pick />
-	<PickHubScript on:pick />
+	<PickScript {isTrigger} on:pick />
+	<PickHubScript {isTrigger} on:pick />
 	<FlowScriptPicker
-		label="New Typescript script (Deno)"
+		label="New Typescript {isTrigger ? 'trigger ' : ''}script (Deno)"
 		icon={faCode}
 		iconColor="text-blue-800"
 		on:click={() => dispatch('new', { language: FlowModuleValue.language.DENO })}
 	/>
 	<FlowScriptPicker
-		label="New Python script (3.10)"
+		label="New Python {isTrigger ? 'trigger ' : ''}script (3.10)"
 		icon={faCode}
 		iconColor="text-yellow-500"
 		on:click={() => dispatch('new', { language: FlowModuleValue.language.PYTHON3 })}
--- a/frontend/src/lib/components/flows/flowStore.ts
+++ b/frontend/src/lib/components/flows/flowStore.ts
@@ -5,6 +5,8 @@ import { userStore, workspaceStore } from '$lib/stores'
 import { derived, get, writable } from 'svelte/store'
 import { createInlineScriptModuleFromPath, getFirstStepSchema, loadSchemaFromModule } from './utils'

+export type FlowMode = 'push' | 'pull'
+
 export const flowStore = writable<Flow>(undefined)
 export const schemasStore = writable<Schema[]>([])

@@ -48,13 +50,13 @@ export async function pickScript(path: string, step: number) {
 	await loadSchema(step)
 }

-export async function createInlineScriptModule(language: FlowModuleValue.language, step: number) {
+export async function createInlineScriptModule(language: FlowModuleValue.language, step: number, mode: FlowMode) {
 	const code = language === FlowModuleValue.language.DENO ? DENO_INIT_CODE : PYTHON_INIT_CODE
 	flowStore.update((flow: Flow) => {
 		flow.value.modules[step].value = {
 			type: FlowModuleValue.type.RAWSCRIPT,
 			content: code,
-			language
+			language,
 		}

 		return flow
--- a/frontend/src/lib/components/flows/pickers/PickHubScript.svelte
+++ b/frontend/src/lib/components/flows/pickers/PickHubScript.svelte
@@ -6,6 +6,8 @@
 	import { hubScripts } from '$lib/stores'
 	import { createEventDispatcher } from 'svelte'

+	export let isTrigger: boolean
+
 	type Item = { summary: String; path: String; version?: String }

 	const items: Item[] = $hubScripts ?? []
--- a/frontend/src/lib/components/flows/pickers/PickScript.svelte
+++ b/frontend/src/lib/components/flows/pickers/PickScript.svelte
@@ -7,6 +7,8 @@
 	import { createEventDispatcher } from 'svelte'
 	import FlowScriptPicker from './FlowScriptPicker.svelte'

+	export let isTrigger: boolean
+
 	type Item = { summary: String; path: String; version?: String }

 	let items: Item[] = []
--- a/frontend/src/lib/components/flows/utils.ts
+++ b/frontend/src/lib/components/flows/utils.ts
@@ -12,6 +12,44 @@ import { DENO_INIT_CODE, PYTHON_INIT_CODE } from '$lib/script_helpers'
 import { workspaceStore } from '$lib/stores'
 import { emptySchema } from '$lib/utils'
 import { get } from 'svelte/store'
+import type { FlowMode } from './flowStore'
+
+export function flowToMode(flow: Flow, mode: FlowMode): Flow {
+	if (mode == 'pull') {
+		const newFlow: Flow = JSON.parse(JSON.stringify(flow))
+		const triggerModule = newFlow.value.modules[0]
+		const oldModules = newFlow.value.modules.slice(1)
+		if (triggerModule) {
+			triggerModule.stop_after_if_expr = "result.res1.length == 0"
+			triggerModule.skip_if_stopped = true
+		}
+		newFlow.value.modules = newFlow.value.modules.slice(0, 1)
+		newFlow.value.modules.push({
+			input_transform: oldModules[0].input_transform,
+			value: {
+				type: FlowModuleValue.type.FORLOOPFLOW,
+				iterator: { type: InputTransform.type.JAVASCRIPT, expr: 'result.res1' },
+				value: {
+					modules: oldModules
+				}
+			}
+		})
+		return newFlow
+	}
+	return flow
+}
+
+export function flattenForloopFlows(flow: Flow): Flow {
+	let newFlow: Flow = JSON.parse(JSON.stringify(flow))
+	if (newFlow.value.modules[1]?.value.type == FlowModuleValue.type.FORLOOPFLOW) {
+		if (newFlow.value.modules.length > 0) {
+			const oldModules = newFlow.value.modules[1].value.value?.modules ?? []
+			newFlow.value.modules = newFlow.value.modules.slice(0, 1)
+			newFlow.value.modules.push(...oldModules)
+		}
+	}
+	return newFlow
+}

 function filterByKey(obj: Object, key: string): Object {
 	if (Object(obj) !== obj) {
--- a/frontend/src/lib/utils.ts
+++ b/frontend/src/lib/utils.ts
@@ -4,6 +4,8 @@ import { toast } from '@zerodevx/svelte-toast'
 import type { Schema } from './common'
 import type { UserExt } from './stores'

+
+
 export function isToday(someDate: Date): boolean {
 	const today = new Date()
 	return (
@@ -39,9 +41,8 @@ export function displayDate(dateString: string | undefined): string {
 	if (date.toString() === 'Invalid Date') {
 		return ''
 	} else {
-		return `${date.getFullYear()}/${
-			date.getMonth() + 1
-		}/${date.getDate()} at ${date.toLocaleTimeString()}`
+		return `${date.getFullYear()}/${date.getMonth() + 1
+			}/${date.getDate()} at ${date.toLocaleTimeString()}`
 	}
 }

--- a/frontend/src/routes/flows/edit/[...path].svelte
+++ b/frontend/src/routes/flows/edit/[...path].svelte
@@ -5,6 +5,7 @@
 	import FlowBuilder from '$lib/components/FlowBuilder.svelte'
 	import { workspaceStore } from '$lib/stores'
 	import { emptySchema } from '$lib/utils'
+	import { flattenForloopFlows } from '$lib/components/flows/utils'

 	const initialState = $page.url.searchParams.get('state')
 	let flowLoadedFromUrl = initialState != undefined ? JSON.parse(atob(initialState)) : undefined
@@ -30,6 +31,7 @@
 						workspace: $workspaceStore!,
 						path: flow.path
 				  })
+		flow = flattenForloopFlows(flow)
 		initialPath = flow.path
 	}

--- a/frontend/src/routes/run/[...run].svelte
+++ b/frontend/src/routes/run/[...run].svelte
@@ -26,7 +26,8 @@
 		faHourglassHalf,
 		faRobot,
 		faScroll,
-		faWind
+		faWind,
+		faFastForward
 	} from '@fortawesome/free-solid-svg-icons'
 	import Tooltip from '$lib/components/Tooltip.svelte'
 	import DisplayResult from '$lib/components/DisplayResult.svelte'
@@ -145,12 +146,21 @@
 			<div>
 				{#if job}
 					{#if 'success' in job && job.success}
-						<Icon
-							class="text-green-600"
-							data={check}
-							scale={iconScale}
-							label="Job completed successfully"
-						/>
+						{#if job.is_skipped}
+							<Icon
+								class="text-green-600"
+								data={faFastForward}
+								scale={SMALL_ICON_SCALE}
+								label="Job completed successfully but was skipped"
+							/>
+						{:else}
+							<Icon
+								class="text-green-600"
+								data={check}
+								scale={SMALL_ICON_SCALE}
+								label="Job completed successfully"
+							/>
+						{/if}
 					{:else if job && 'success' in job}
 						<Icon
 							class="text-red-700"
--- a/frontend/src/routes/runs/[...path].svelte
+++ b/frontend/src/routes/runs/[...path].svelte
@@ -8,6 +8,7 @@
 		faCalendar,
 		faCircle,
 		faClock,
+		faFastForward,
 		faHourglassHalf,
 		faRobot,
 		faTimes,
@@ -130,8 +131,14 @@ the bearer token they use has less privilege."
 		<div class="xl:max-w-screen-lg">
 			<Tabs
 				tabs={[
-					['script,dependencies,preview,flow', 'all'],
-					[`${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW}`, 'runs'],
+					[
+						`${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.PREVIEW},${CompletedJob.job_kind.FLOWPREVIEW},${CompletedJob.job_kind.SCRIPT_HUB}`,
+						'all'
+					],
+					[
+						`${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.SCRIPT_HUB}`,
+						'runs'
+					],
 					[`${CompletedJob.job_kind.PREVIEW},${CompletedJob.job_kind.FLOWPREVIEW}`, 'previews'],
 					[CompletedJob.job_kind.DEPENDENCIES, 'dependencies']
 				]}
@@ -151,12 +158,21 @@ the bearer token they use has less privilege."
 									{:else}
 										<div class="block text-center align-middle pb-3 pt-2 px-6">
 											{#if 'success' in job && job.success}
-												<Icon
-													class="text-green-600"
-													data={check}
-													scale={SMALL_ICON_SCALE}
-													label="Job completed successfully"
-												/>
+												{#if job.is_skipped}
+													<Icon
+														class="text-green-600"
+														data={faFastForward}
+														scale={SMALL_ICON_SCALE}
+														label="Job completed successfully but was skipped"
+													/>
+												{:else}
+													<Icon
+														class="text-green-600"
+														data={check}
+														scale={SMALL_ICON_SCALE}
+														label="Job completed successfully"
+													/>
+												{/if}
 											{:else if 'success' in job}
 												<Icon
 													class="text-red-700"