feat: Restartable flows (#2514)

* feat: Restartable flows for simple sequential flows

* check if status of previous module to initialize last_result

* compute result of forloop flows when needed

* Add possibility to restart a flow preview

* Re-compute leaf jobs when needed

* FE polishing

* Add tooltip on button and fix module not found error in preview

* Handle ListJob when computing leaf job from db

* Fix unused imports

* ./build_openapi.sh

* sqlx prepare

backend/.sqlx/query-061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flow_status",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168"
+}

backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json

@@ -42,7 +42,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json

@@ -67,7 +67,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json

@@ -28,7 +28,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json

@@ -60,7 +60,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json

@@ -40,7 +40,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json

@@ -46,7 +46,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json

@@ -42,7 +42,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json

@@ -42,7 +42,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json

@@ -37,7 +37,6 @@
                 "bash",
                 "postgresql",
                 "nativets",
-                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/tests/worker.rs

@@ -276,7 +276,7 @@ mod suspend_resume {
         let server = ApiServer::start(db.clone()).await;
         let port = server.addr.port();
 
-        let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
+        let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None })
             .arg("n", json!(1))
             .arg("port", json!(port))
             .push(&db)
@@ -357,7 +357,7 @@ mod suspend_resume {
         let server = ApiServer::start(db.clone()).await;
         let port = server.addr.port();
 
-        let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
+        let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None })
             .arg("n", json!(1))
             .arg("op", json!("cancel"))
             .arg("port", json!(port))
@@ -381,7 +381,7 @@ mod suspend_resume {
         let server = ApiServer::start(db.clone()).await;
         let port = server.addr.port();
 
-        let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
+        let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None })
             .arg("n", json!(1))
             .arg("port", json!(port))
             .push(&db)
@@ -576,7 +576,7 @@ def main(last, port):
         .into_iter()
         .unzip::<_, _, Vec<_>, Vec<_>>();
         let server = Server::start(responses).await;
-        let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None })
+        let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None })
             .arg("items", json!(["unused", "unused", "unused"]))
             .arg("port", json!(server.addr.port()))
             .run_until_complete(&db, server.addr.port())
@@ -605,7 +605,7 @@ def main(last, port):
         .into_iter()
         .unzip::<_, _, Vec<_>, Vec<_>>();
         let server = Server::start(responses).await;
-        let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None })
+        let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None })
             .arg("items", json!(["unused", "unused", "unused"]))
             .arg("port", json!(server.addr.port()))
             .run_until_complete(&db, server.addr.port())
@@ -646,7 +646,7 @@ def main(last, port):
         .into_iter()
         .unzip::<_, _, Vec<_>, Vec<_>>();
         let server = Server::start(responses).await;
-        let job = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None })
+        let job = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None })
             .arg("items", json!(["unused", "unused", "unused"]))
             .arg("port", json!(server.addr.port()))
             .run_until_complete(&db, server.addr.port())
@@ -712,7 +712,7 @@ def main(error, port):
         .into_iter()
         .unzip::<_, _, Vec<_>, Vec<_>>();
         let server = Server::start(responses).await;
-        let cjob = RunJob::from(JobPayload::RawFlow { value, path: None })
+        let cjob = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None })
             .arg("port", json!(server.addr.port()))
             .run_until_complete(&db, server.addr.port())
             .await;
@@ -770,7 +770,7 @@ async fn test_iteration(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("items", json!([]))
         .run_until_complete(&db, server.addr.port())
         .await
@@ -779,7 +779,7 @@ async fn test_iteration(db: Pool<Postgres>) {
     assert_eq!(result, serde_json::json!([]));
 
     /* Don't actually test that this does 257 jobs or that will take forever. */
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("items", json!((0..257).collect::<Vec<_>>()))
         .run_until_complete(&db, server.addr.port())
         .await
@@ -827,7 +827,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("items", json!([]))
         .run_until_complete(&db, server.addr.port())
         .await
@@ -836,7 +836,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
     assert_eq!(result, serde_json::json!([]));
 
     /* Don't actually test that this does 257 jobs or that will take forever. */
-    let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("items", json!((0..50).collect::<Vec<_>>()))
         .run_until_complete(&db, server.addr.port())
         .await;
@@ -1136,7 +1136,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
         }
     };
 
-    let job = JobPayload::RawFlow { value: flow, path: None };
+    let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let port = server.addr.port();
 
     for i in 0..50 {
@@ -1176,7 +1176,7 @@ async fn test_identity(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .run_until_complete(&db, server.addr.port())
         .await
         .json_result()
@@ -1365,7 +1365,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
             ..Default::default()
         };
 
-    let job = JobPayload::RawFlow { value: flow, path: None };
+    let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
 
     let result = run_job_in_new_worker_until_complete(&db, job.clone(), server.addr.port())
         .await
@@ -1421,7 +1421,7 @@ async fn test_flow_result_by_id(db: Pool<Postgres>) {
         }))
         .unwrap();
 
-    let job = JobPayload::RawFlow { value: flow, path: None };
+    let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, job.clone(), port)
         .await
         .json_result()
@@ -1463,7 +1463,7 @@ async fn test_stop_after_if(db: Pool<Postgres>) {
         ],
     }))
     .unwrap();
-    let job = JobPayload::RawFlow { value: flow, path: None };
+    let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
 
     let result = RunJob::from(job.clone())
         .arg("n", json!(123))
@@ -1521,7 +1521,7 @@ async fn test_stop_after_if_nested(db: Pool<Postgres>) {
         ],
     }))
     .unwrap();
-    let job = JobPayload::RawFlow { value: flow, path: None };
+    let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
 
     let result = RunJob::from(job.clone())
         .arg("n", json!(123))
@@ -1586,7 +1586,7 @@ async fn test_python_flow(db: Pool<Postgres>) {
         println!("python flow iteration: {}", i);
         let result = run_job_in_new_worker_until_complete(
             &db,
-            JobPayload::RawFlow { value: flow.clone(), path: None },
+            JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None },
             port,
         )
         .await
@@ -1621,7 +1621,7 @@ async fn test_python_flow_2(db: Pool<Postgres>) {
         println!("python flow iteration: {}", i);
         let result = run_job_in_new_worker_until_complete(
             &db,
-            JobPayload::RawFlow { value: flow.clone(), path: None },
+            JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None },
             port,
         )
         .await
@@ -1837,7 +1837,7 @@ async fn test_empty_loop(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -1876,7 +1876,7 @@ async fn test_invalid_first_step(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let job = run_job_in_new_worker_until_complete(&db, flow, port).await;
 
     assert_eq!(
@@ -1918,7 +1918,7 @@ async fn test_empty_loop_2(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -1973,7 +1973,7 @@ async fn test_step_after_loop(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2041,7 +2041,7 @@ async fn test_branchone_simple(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2077,7 +2077,7 @@ async fn test_branchone_with_cond(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2115,7 +2115,7 @@ async fn test_branchall_sequential(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2152,7 +2152,7 @@ async fn test_branchall_simple(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2199,7 +2199,7 @@ async fn test_branchall_skip_failure(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2236,7 +2236,7 @@ async fn test_branchall_skip_failure(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2300,7 +2300,7 @@ async fn test_branchone_nested(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2357,7 +2357,7 @@ async fn test_branchall_nested(db: Pool<Postgres>) {
     }))
     .unwrap();
 
-    let flow = JobPayload::RawFlow { value: flow, path: None };
+    let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
     let result = run_job_in_new_worker_until_complete(&db, flow, port)
         .await
         .json_result()
@@ -2421,7 +2421,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
         }))
         .unwrap();
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("n", json!(0))
         .run_until_complete(&db, port)
         .await
@@ -2437,7 +2437,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
         .unwrap()
         .contains("[]"));
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("n", json!(1))
         .run_until_complete(&db, port)
         .await
@@ -2453,7 +2453,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
         .unwrap()
         .contains("[0]"));
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("n", json!(2))
         .run_until_complete(&db, port)
         .await
@@ -2469,7 +2469,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
         .unwrap()
         .contains("[0,1]"));
 
-    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
+    let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
         .arg("n", json!(3))
         .run_until_complete(&db, port)
         .await

backend/windmill-api/openapi-deref.yaml

@@ -1,6 +1,6 @@
 openapi: 3.0.3
 info:
-  version: 1.188.1
+  version: 1.192.0
   title: Windmill API
   contact:
     name: Windmill Team
@@ -2329,7 +2329,7 @@ paths:
             application/json:
               schema:
                 type: object
-                properties: &ref_151
+                properties: &ref_152
                   access_token:
                     type: string
                   expires_in:
@@ -2340,7 +2340,7 @@ paths:
                     type: array
                     items:
                       type: string
-                required: &ref_152
+                required: &ref_153
                   - access_token
   /w/{workspace}/oauth/create_account:
     post:
@@ -3189,32 +3189,32 @@ paths:
                                 id:
                                   type: string
                                 value:
-                                  oneOf: &ref_173
+                                  oneOf: &ref_174
                                     - type: object
-                                      properties: &ref_157
+                                      properties: &ref_158
                                         input_transforms:
                                           type: object
                                           additionalProperties:
                                             oneOf: &ref_23
                                               - type: object
-                                                properties: &ref_153
+                                                properties: &ref_154
                                                   value: {}
                                                   type:
                                                     type: string
                                                     enum:
                                                       - javascript
-                                                required: &ref_154
+                                                required: &ref_155
                                                   - expr
                                                   - type
                                               - type: object
-                                                properties: &ref_155
+                                                properties: &ref_156
                                                   expr:
                                                     type: string
                                                   type:
                                                     type: string
                                                     enum:
                                                       - javascript
-                                                required: &ref_156
+                                                required: &ref_157
                                                   - expr
                                                   - type
                                             discriminator: &ref_24
@@ -3253,13 +3253,13 @@ paths:
                                           type: number
                                         concurrency_time_window_s:
                                           type: number
-                                      required: &ref_158
+                                      required: &ref_159
                                         - type
                                         - content
                                         - language
                                         - input_transforms
                                     - type: object
-                                      properties: &ref_159
+                                      properties: &ref_160
                                         input_transforms:
                                           type: object
                                           additionalProperties:
@@ -3273,12 +3273,12 @@ paths:
                                           type: string
                                           enum:
                                             - script
-                                      required: &ref_160
+                                      required: &ref_161
                                         - type
                                         - path
                                         - input_transforms
                                     - type: object
-                                      properties: &ref_161
+                                      properties: &ref_162
                                         input_transforms:
                                           type: object
                                           additionalProperties:
@@ -3290,12 +3290,12 @@ paths:
                                           type: string
                                           enum:
                                             - flow
-                                      required: &ref_162
+                                      required: &ref_163
                                         - type
                                         - path
                                         - input_transforms
                                     - type: object
-                                      properties: &ref_163
+                                      properties: &ref_164
                                         modules:
                                           type: array
                                           items:
@@ -3317,13 +3317,13 @@ paths:
                                           type: boolean
                                         parallelism:
                                           type: integer
-                                      required: &ref_164
+                                      required: &ref_165
                                         - modules
                                         - iterator
                                         - skip_failures
                                         - type
                                     - type: object
-                                      properties: &ref_165
+                                      properties: &ref_166
                                         branches:
                                           type: array
                                           items:
@@ -3354,12 +3354,12 @@ paths:
                                           type: string
                                           enum:
                                             - branchone
-                                      required: &ref_166
+                                      required: &ref_167
                                         - branches
                                         - default
                                         - type
                                     - type: object
-                                      properties: &ref_167
+                                      properties: &ref_168
                                         branches:
                                           type: array
                                           items:
@@ -3384,28 +3384,28 @@ paths:
                                             - branchall
                                         parallel:
                                           type: boolean
-                                      required: &ref_168
+                                      required: &ref_169
                                         - branches
                                         - type
                                     - type: object
-                                      properties: &ref_169
+                                      properties: &ref_170
                                         type:
                                           type: string
                                           enum:
                                             - identity
                                         flow:
                                           type: boolean
-                                      required: &ref_170
+                                      required: &ref_171
                                         - type
                                     - type: object
-                                      properties: &ref_171
+                                      properties: &ref_172
                                         type:
                                           type: string
                                           enum:
                                             - graphql
-                                      required: &ref_172
+                                      required: &ref_173
                                         - type
-                                  discriminator: &ref_174
+                                  discriminator: &ref_175
                                     propertyName: type
                                     mapping:
                                       rawscript: '#/components/schemas/RawScript'
@@ -3461,7 +3461,7 @@ paths:
                                   type: number
                                 retry:
                                   type: object
-                                  properties: &ref_175
+                                  properties: &ref_176
                                     constant:
                                       type: object
                                       properties:
@@ -3981,13 +3981,9 @@ paths:
                       items:
                         type: string
                     concurrent_limit:
-                      type: array
-                      items:
-                        type: integer
+                      type: integer
                     concurrency_time_window_s:
-                      type: array
-                      items:
-                        type: integer
+                      type: integer
                     cache_ttl:
                       type: number
                     dedicated_worker:
@@ -4169,13 +4165,9 @@ paths:
                   items:
                     type: string
                 concurrent_limit:
-                  type: array
-                  items:
-                    type: integer
+                  type: integer
                 concurrency_time_window_s:
-                  type: array
-                  items:
-                    type: integer
+                  type: integer
                 cache_ttl:
                   type: number
                 dedicated_worker:
@@ -5411,7 +5403,7 @@ paths:
                 type: array
                 items:
                   type: object
-                  properties: &ref_148
+                  properties: &ref_149
                     workspace_id:
                       type: string
                     path:
@@ -5429,7 +5421,7 @@ paths:
                     edited_at:
                       type: string
                       format: date-time
-                  required: &ref_149
+                  required: &ref_150
                     - workspace_id
                     - path
                     - summary
@@ -5565,7 +5557,7 @@ paths:
                 type: array
                 items:
                   type: object
-                  properties: &ref_146
+                  properties: &ref_147
                     id:
                       type: integer
                     workspace_id:
@@ -5591,7 +5583,7 @@ paths:
                         - viewer
                         - publisher
                         - anonymous
-                  required: &ref_147
+                  required: &ref_148
                     - id
                     - workspace_id
                     - path
@@ -5764,7 +5756,7 @@ paths:
           content:
             application/json:
               schema:
-                allOf: &ref_150
+                allOf: &ref_151
                   - type: object
                     properties: *ref_45
                     required: *ref_46
@@ -6142,6 +6134,84 @@ paths:
               schema:
                 type: string
                 format: uuid
+  /w/{workspace}/jobs/restart/f/{id}/from/{step_id}:
+    post:
+      summary: restart a completed flow at a given step
+      operationId: restartFlowAtStep
+      tags:
+        - job
+      parameters:
+        - name: workspace
+          in: path
+          required: true
+          schema: *ref_0
+        - name: id
+          in: path
+          required: true
+          schema: &ref_69
+            type: string
+            format: uuid
+        - name: step_id
+          description: step id to restart the flow from
+          required: true
+          in: path
+          schema:
+            type: string
+        - name: scheduled_for
+          description: when to schedule this job (leave empty for immediate run)
+          in: query
+          schema:
+            type: string
+            format: date-time
+        - name: scheduled_in_secs
+          description: schedule the script to execute in the number of seconds starting now
+          in: query
+          schema:
+            type: integer
+        - name: parent_job
+          description: >-
+            The parent job that is at the origin and responsible for the
+            execution of this script if any
+          in: query
+          schema: *ref_34
+        - name: job_id
+          description: >-
+            The job id to assign to the created job. if missing, job is chosen
+            randomly using the ULID scheme. If a job id already exists in the
+            queue or as a completed job, the request to create one will fail
+            (Bad Request)
+          in: query
+          schema: *ref_35
+        - name: include_header
+          description: >
+            List of headers's keys (separated with ',') whove value are added to
+            the args
+
+            Header's key lowercased and '-'' replaced to '_' such that
+            'Content-Type' becomes the 'content_type' arg key
+          in: query
+          schema: *ref_36
+        - name: invisible_to_owner
+          description: make the run invisible to the the flow owner (default false)
+          in: query
+          schema:
+            type: boolean
+      requestBody:
+        description: flow args
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: *ref_14
+      responses:
+        '201':
+          description: job created
+          content:
+            text/plain:
+              schema:
+                type: string
+                format: uuid
   /w/{workspace}/jobs/run/h/{hash}:
     post:
       summary: run script by hash
@@ -6344,6 +6414,14 @@ paths:
                   additionalProperties: *ref_14
                 tag:
                   type: string
+                restarted_from:
+                  type: object
+                  properties: &ref_146
+                    flow_job_id:
+                      type: string
+                      format: uuid
+                    step_id:
+                      type: string
               required: &ref_145
                 - value
                 - content
@@ -7049,7 +7127,7 @@ paths:
               schema:
                 type: array
                 items:
-                  allOf: &ref_69
+                  allOf: &ref_70
                     - oneOf:
                         - type: object
                           properties: *ref_65
@@ -7064,7 +7142,7 @@ paths:
                           enum:
                             - CompletedJob
                             - QueuedJob
-                  discriminator: &ref_70
+                  discriminator: &ref_71
                     propertyName: type
   /jobs/db_clock:
     get:
@@ -7093,17 +7171,15 @@ paths:
         - name: id
           in: path
           required: true
-          schema: &ref_71
-            type: string
-            format: uuid
+          schema: *ref_69
       responses:
         '200':
           description: job details
           content:
             application/json:
               schema:
-                allOf: *ref_69
-                discriminator: *ref_70
+                allOf: *ref_70
+                discriminator: *ref_71
   /w/{workspace}/jobs_u/get_logs/{id}:
     get:
       summary: get job logs
@@ -7118,7 +7194,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       responses:
         '200':
           description: job details
@@ -7140,7 +7216,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: running
           in: query
           schema:
@@ -7179,7 +7255,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       responses:
         '200':
           description: job details
@@ -7203,7 +7279,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       responses:
         '200':
           description: result
@@ -7224,7 +7300,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: get_started
           in: query
           schema: &ref_98
@@ -7259,7 +7335,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       responses:
         '200':
           description: job details
@@ -7283,7 +7359,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       requestBody:
         description: reason
         required: true
@@ -7315,7 +7391,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       requestBody:
         description: reason
         required: true
@@ -7347,7 +7423,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7378,7 +7454,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7420,7 +7496,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: payload
           description: >
             The base64 encoded payload that has been encoded as a JSON. e.g how
@@ -7463,7 +7539,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7505,7 +7581,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
       requestBody:
         required: true
         content:
@@ -7533,7 +7609,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7568,7 +7644,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7610,7 +7686,7 @@ paths:
         - name: id
           in: path
           required: true
-          schema: *ref_71
+          schema: *ref_69
         - name: resume_id
           in: path
           required: true
@@ -7634,8 +7710,8 @@ paths:
                 type: object
                 properties:
                   job:
-                    allOf: *ref_69
-                    discriminator: *ref_70
+                    allOf: *ref_70
+                    discriminator: *ref_71
                   approvers:
                     type: array
                     items:
@@ -9249,7 +9325,7 @@ components:
       name: id
       in: path
       required: true
-      schema: *ref_71
+      schema: *ref_69
     Path:
       name: path
       in: path
@@ -9492,8 +9568,8 @@ components:
       properties: *ref_65
       required: *ref_66
     Job:
-      allOf: *ref_69
-      discriminator: *ref_70
+      allOf: *ref_70
+      discriminator: *ref_71
     User:
       type: object
       properties: *ref_8
@@ -9747,23 +9823,26 @@ components:
       type: object
       properties: *ref_144
       required: *ref_145
+    RestartedFrom:
+      type: object
+      properties: *ref_146
     Policy:
       type: object
       properties: *ref_44
     ListableApp:
       type: object
-      properties: *ref_146
-      required: *ref_147
+      properties: *ref_147
+      required: *ref_148
     ListableRawApp:
       type: object
-      properties: *ref_148
-      required: *ref_149
+      properties: *ref_149
+      required: *ref_150
     AppWithLastVersion:
       type: object
       properties: *ref_45
       required: *ref_46
     AppWithLastVersionWDraft:
-      allOf: *ref_150
+      allOf: *ref_151
     SlackToken:
       type: object
       properties:
@@ -9785,64 +9864,64 @@ components:
         - bot
     TokenResponse:
       type: object
-      properties: *ref_151
-      required: *ref_152
+      properties: *ref_152
+      required: *ref_153
     HubScriptKind:
       name: kind
       schema: *ref_28
     StaticTransform:
       type: object
-      properties: *ref_153
-      required: *ref_154
+      properties: *ref_154
+      required: *ref_155
     JavascriptTransform:
       type: object
-      properties: *ref_155
-      required: *ref_156
+      properties: *ref_156
+      required: *ref_157
     InputTransform:
       oneOf: *ref_23
       discriminator: *ref_24
     RawScript:
       type: object
-      properties: *ref_157
-      required: *ref_158
+      properties: *ref_158
+      required: *ref_159
     PathScript:
       type: object
-      properties: *ref_159
-      required: *ref_160
+      properties: *ref_160
+      required: *ref_161
     PathFlow:
       type: object
-      properties: *ref_161
-      required: *ref_162
+      properties: *ref_162
+      required: *ref_163
     FlowModule:
       type: object
       properties: *ref_25
       required: *ref_26
     ForloopFlow:
       type: object
-      properties: *ref_163
-      required: *ref_164
+      properties: *ref_164
+      required: *ref_165
     BranchOne:
       type: object
-      properties: *ref_165
-      required: *ref_166
+      properties: *ref_166
+      required: *ref_167
     BranchAll:
       type: object
-      properties: *ref_167
-      required: *ref_168
+      properties: *ref_168
+      required: *ref_169
     Identity:
       type: object
-      properties: *ref_169
-      required: *ref_170
+      properties: *ref_170
+      required: *ref_171
     Graphql:
       type: object
-      properties: *ref_171
-      required: *ref_172
+      properties: *ref_172
+      required: *ref_173
     FlowModuleValue:
-      oneOf: *ref_173
-      discriminator: *ref_174
+      oneOf: *ref_174
+      discriminator: *ref_175
     Retry:
       type: object
-      properties: *ref_175
+      properties: *ref_176
     FlowValue:
       type: object
       properties: *ref_47

backend/windmill-api/openapi.yaml

@@ -4229,6 +4229,58 @@ paths:
                 type: string
                 format: uuid
 
+  /w/{workspace}/jobs/restart/f/{id}/from/{step_id}:
+    post:
+      summary: restart a completed flow at a given step
+      operationId: restartFlowAtStep
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/JobId"
+        - name: step_id
+          description: step id to restart the flow from
+          required: true
+          in: path
+          schema:
+            type: string
+        - name: scheduled_for
+          description: when to schedule this job (leave empty for immediate run)
+          in: query
+          schema:
+            type: string
+            format: date-time
+        - name: scheduled_in_secs
+          description: schedule the script to execute in the number of seconds starting now
+          in: query
+          schema:
+            type: integer
+        - $ref: "#/components/parameters/ParentJob"
+        - $ref: "#/components/parameters/NewJobId"
+        - $ref: "#/components/parameters/IncludeHeader"
+        - name: invisible_to_owner
+          description: make the run invisible to the the flow owner (default false)
+          in: query
+          schema:
+            type: boolean
+
+      requestBody:
+        description: flow args
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/ScriptArgs"
+
+      responses:
+        "201":
+          description: job created
+          content:
+            text/plain:
+              schema:
+                type: string
+                format: uuid
+
   /w/{workspace}/jobs/run/h/{hash}:
     post:
       summary: run script by hash
@@ -7865,12 +7917,23 @@ components:
           $ref: "#/components/schemas/ScriptArgs"
         tag:
           type: string
+        restarted_from:
+          $ref: "#/components/schemas/RestartedFrom"
 
       required:
         - value
         - content
         - args
 
+    RestartedFrom:
+      type: object
+      properties:
+        flow_job_id:
+          type: string
+          format: uuid
+        step_id:
+          type: string
+
     Policy:
       type: object
       properties:

backend/windmill-api/src/jobs.rs

@@ -8,6 +8,7 @@
 
 use serde_json::value::RawValue;
 use std::collections::HashMap;
+use windmill_common::flow_status::RestartedFrom;
 
 use crate::db::ApiAuthed;
 
@@ -41,7 +42,7 @@ use windmill_common::{
     error::{self, to_anyhow, Error},
     flow_status::{Approval, FlowStatus, FlowStatusModule},
     flows::FlowValue,
-    jobs::{script_path_to_payload, JobKind, JobPayload, QueuedJob, RawCode},
+    jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
     oauth2::HmacSha256,
     scripts::{ScriptHash, ScriptLang},
     users::username_to_permissioned_as,
@@ -63,6 +64,10 @@ pub fn workspaced_service() -> Router {
                 .head(|| async { "" })
                 .layer(cors.clone()),
         )
+        .route(
+            "/restart/f/:job_id/from/*step_id",
+            post(restart_flow).head(|| async { "" }).layer(cors.clone()),
+        )
         .route(
             "/run/p/*script_path",
             post(run_job_by_path)
@@ -398,64 +403,6 @@ async fn get_job_logs(
     Ok(text)
 }
 
-#[derive(Debug, sqlx::FromRow, Serialize)]
-pub struct CompletedJob {
-    pub workspace_id: String,
-    pub id: Uuid,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub parent_job: Option<Uuid>,
-    pub created_by: String,
-    pub created_at: chrono::DateTime<chrono::Utc>,
-    pub started_at: chrono::DateTime<chrono::Utc>,
-    pub duration_ms: i64,
-    pub success: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub script_hash: Option<ScriptHash>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub script_path: Option<String>,
-    pub args: Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub result: Option<sqlx::types::Json<Box<RawValue>>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub logs: Option<String>,
-    pub deleted: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub raw_code: Option<String>,
-    pub canceled: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub canceled_by: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub canceled_reason: Option<String>,
-    pub job_kind: JobKind,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub schedule_path: Option<String>,
-    pub permissioned_as: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
-    pub is_flow_step: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub language: Option<ScriptLang>,
-    pub is_skipped: bool,
-    pub email: String,
-    pub visible_to_owner: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub mem_peak: Option<i32>,
-    pub tag: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub priority: Option<i16>,
-}
-
-impl CompletedJob {
-    pub fn json_result(&self) -> Option<serde_json::Value> {
-        self.result
-            .as_ref()
-            .map(|r| serde_json::from_str(r.get()).ok())
-            .flatten()
-    }
-}
-
 #[derive(Debug, sqlx::FromRow, Serialize)]
 pub struct ListableCompletedJob {
     pub r#type: String,
@@ -501,12 +448,6 @@ pub struct ListableCompletedJob {
     pub priority: Option<i16>,
 }
 
-impl<'a> IntoResponse for CompletedJob {
-    fn into_response(self) -> Response {
-        Json(self).into_response()
-    }
-}
-
 #[derive(Deserialize, Clone)]
 pub struct RunJobQuery {
     scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
@@ -1531,6 +1472,7 @@ struct PreviewFlow {
     path: Option<String>,
     args: Option<Box<JsonRawValue>>,
     tag: Option<String>,
+    restarted_from: Option<RestartedFrom>,
 }
 
 pub struct QueryOrBody<D>(pub Option<D>);
@@ -1686,6 +1628,73 @@ pub async fn run_flow_by_path(
     Ok((StatusCode::CREATED, uuid.to_string()))
 }
 
+pub async fn restart_flow(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
+    Path((w_id, job_id, step_id)): Path<(String, Uuid, String)>,
+    Query(run_query): Query<RunJobQuery>,
+) -> error::Result<(StatusCode, String)> {
+    #[cfg(not(feature = "enterprise"))]
+    {
+        return Err(Error::BadRequest(
+            "Restarting a flow is a feature only available in enterprise version".to_string(),
+        ));
+    }
+
+    #[cfg(feature = "enterprise")]
+    check_license_key_valid().await?;
+
+    let completed_job = sqlx::query_as::<_, CompletedJob>(
+        "SELECT * from completed_job WHERE id = $1 and workspace_id = $2",
+    )
+    .bind(job_id)
+    .bind(&w_id)
+    .fetch_optional(&db)
+    .await?
+    .with_context(|| "Unable to find completed job with the given job UUID")?;
+
+    let flow_path = completed_job
+        .script_path
+        .with_context(|| "No flow path set for completed flow job")?;
+    check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
+
+    let push_args = completed_job
+        .args
+        .map(|json| PushArgs { args: json.clone(), extra: json.0 });
+
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
+
+    let (uuid, tx) = push(
+        &db,
+        tx,
+        &w_id,
+        JobPayload::RestartedFlow { completed_job_id: job_id, step_id: step_id },
+        push_args,
+        &authed.username,
+        &authed.email,
+        username_to_permissioned_as(&authed.username),
+        scheduled_for,
+        None,
+        run_query.parent_job,
+        run_query.parent_job,
+        run_query.job_id,
+        false,
+        false,
+        None,
+        !run_query.invisible_to_owner.unwrap_or(false),
+        Some(completed_job.tag),
+        None,
+        None,
+        completed_job.priority,
+    )
+    .await?;
+    tx.commit().await?;
+    Ok((StatusCode::CREATED, uuid.to_string()))
+}
+
 pub async fn run_job_by_path(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -2324,7 +2333,7 @@ async fn add_batch_jobs(
 
             let mut uuids: Vec<Uuid> = Vec::new();
             let payload = if let Some(ref fv) = batch_info.flow_value {
-                JobPayload::RawFlow { value: fv.clone(), path: None }
+                JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None }
             } else {
                 if let Some(path) = batch_info.path.as_ref() {
                     JobPayload::Flow(path.to_string())
@@ -2442,7 +2451,11 @@ async fn run_preview_flow_job(
         &db,
         tx,
         &w_id,
-        JobPayload::RawFlow { value: raw_flow.value, path: raw_flow.path },
+        JobPayload::RawFlow {
+            value: raw_flow.value,
+            path: raw_flow.path,
+            restarted_from: raw_flow.restarted_from,
+        },
         raw_flow.args.unwrap_or_default(),
         &authed.username,
         &authed.email,
@@ -2769,7 +2782,8 @@ async fn get_completed_job<'a>(
         .await?;
 
     let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;
-    Ok(CompletedJob::from_row(&job)?.into_response())
+    let response = Json(CompletedJob::from_row(&job)?).into_response();
+    Ok(response)
 }
 
 #[derive(FromRow)]
@@ -2900,5 +2914,6 @@ async fn delete_completed_job<'a>(
     .await?;
 
     tx.commit().await?;
-    Ok(CompletedJob::from_row(&job)?.into_response())
+    let response = Json(CompletedJob::from_row(&job)?).into_response();
+    Ok(response)
 }

backend/windmill-common/src/flow_status.rs

@@ -34,6 +34,8 @@ pub struct FlowStatus {
     pub retry: RetryStatus,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub approval_conditions: Option<ApprovalConditions>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub restarted_from: Option<RestartedFrom>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
@@ -50,6 +52,13 @@ pub struct ApprovalConditions {
     pub user_groups_required: Vec<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+#[serde(default)]
+pub struct RestartedFrom {
+    pub flow_job_id: Uuid,
+    pub step_id: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Iterator {
     pub index: usize,
@@ -179,6 +188,13 @@ impl FlowStatusModule {
             FlowStatusModule::Failure { id, .. } => id.clone(),
         }
     }
+
+    pub fn is_failure(&self) -> bool {
+        match self {
+            FlowStatusModule::Failure { .. } => true,
+            _ => false,
+        }
+    }
 }
 
 impl FlowStatus {
@@ -202,6 +218,7 @@ impl FlowStatus {
                 },
             },
             retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
+            restarted_from: None,
         }
     }
 

backend/windmill-common/src/jobs.rs

@@ -7,7 +7,7 @@ use uuid::Uuid;
 
 use crate::{
     error::{self, Error},
-    flow_status::FlowStatus,
+    flow_status::{FlowStatus, RestartedFrom},
     flows::FlowValue,
     get_latest_deployed_hash_for_path,
     scripts::{ScriptHash, ScriptLang},
@@ -186,6 +186,82 @@ impl Default for QueuedJob {
     }
 }
 
+#[derive(Debug, sqlx::FromRow, Serialize, Clone)]
+pub struct CompletedJob {
+    pub workspace_id: String,
+    pub id: Uuid,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub parent_job: Option<Uuid>,
+    pub created_by: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    pub started_at: chrono::DateTime<chrono::Utc>,
+    pub duration_ms: i64,
+    pub success: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub script_hash: Option<ScriptHash>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub script_path: Option<String>,
+    pub args: Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<sqlx::types::Json<Box<RawValue>>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub logs: Option<String>,
+    pub deleted: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub raw_code: Option<String>,
+    pub canceled: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub canceled_by: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub canceled_reason: Option<String>,
+    pub job_kind: JobKind,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub schedule_path: Option<String>,
+    pub permissioned_as: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
+    pub is_flow_step: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub language: Option<ScriptLang>,
+    pub is_skipped: bool,
+    pub email: String,
+    pub visible_to_owner: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub mem_peak: Option<i32>,
+    pub tag: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub priority: Option<i16>,
+}
+
+impl CompletedJob {
+    pub fn json_result(&self) -> Option<serde_json::Value> {
+        self.result
+            .as_ref()
+            .map(|r| serde_json::from_str(r.get()).ok())
+            .flatten()
+    }
+
+    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
+        self.raw_flow
+            .as_ref()
+            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
+    }
+
+    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
+        self.flow_status
+            .as_ref()
+            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+    }
+}
+
+#[derive(sqlx::FromRow)]
+pub struct BranchResults<'a> {
+    pub result: &'a RawValue,
+    pub id: Uuid,
+}
+
 #[derive(Debug, Clone)]
 pub enum JobPayload {
     ScriptHub {
@@ -216,9 +292,14 @@ pub enum JobPayload {
         version: i64,
     },
     Flow(String),
+    RestartedFlow {
+        completed_job_id: Uuid,
+        step_id: String,
+    },
     RawFlow {
         value: FlowValue,
         path: Option<String>,
+        restarted_from: Option<RestartedFrom>,
     },
     Identity,
     Noop,

backend/windmill-queue/src/jobs.rs

@@ -6,7 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use std::{collections::HashMap, sync::Arc, vec};
+use std::{collections::HashMap, iter, sync::Arc, vec};
 
 use anyhow::Context;
 use async_recursion::async_recursion;
@@ -19,6 +19,7 @@ use axum::{
 };
 use bigdecimal::ToPrimitive;
 use chrono::{DateTime, Duration, Utc};
+use itertools::Itertools;
 use prometheus::IntCounter;
 use reqwest::{
     header::{HeaderMap, CONTENT_TYPE},
@@ -41,12 +42,13 @@ use windmill_common::{
     db::{Authed, UserDB},
     error::{self, Error},
     flow_status::{
-        FlowStatus, FlowStatusModule, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
+        FlowStatus, FlowStatusModule, FlowStatusModuleWParent, JobResult, RestartedFrom,
+        RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
     flows::{FlowModule, FlowModuleValue, FlowValue},
     jobs::{
-        get_payload_tag_from_prefixed_path, script_path_to_payload, JobKind, JobPayload, QueuedJob,
-        RawCode,
+        get_payload_tag_from_prefixed_path, script_path_to_payload, CompletedJob, JobKind,
+        JobPayload, QueuedJob, RawCode,
     },
     schedule::{schedule_to_user, Schedule},
     scripts::{ScriptHash, ScriptLang},
@@ -1467,13 +1469,60 @@ struct ResultR {
     result: Option<Json<Box<RawValue>>>,
 }
 
-#[async_recursion]
 pub async fn get_result_by_id(
     db: Pool<Postgres>,
     w_id: String,
     flow_id: Uuid,
     node_id: String,
     json_path: Option<String>,
+) -> error::Result<Box<RawValue>> {
+    match get_result_by_id_from_running_flow(
+        &db,
+        w_id.as_str(),
+        &flow_id,
+        node_id.as_str(),
+        json_path.clone(),
+    )
+    .await
+    {
+        Ok(res) => Ok(res),
+        Err(_) => {
+            let running_flow_job = sqlx::query_as::<_, QueuedJob>(
+                "SELECT * FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
+            ).bind(flow_id)
+            .bind(&w_id)
+            .fetch_optional(&db)
+            .await?;
+
+            let restarted_from = windmill_common::utils::not_found_if_none(
+                running_flow_job
+                    .map(|fj| fj.parse_flow_status())
+                    .flatten()
+                    .map(|status| status.restarted_from)
+                    .flatten(),
+                "Flow result by id in leaf jobs",
+                format!("{}, {}", flow_id, node_id),
+            )?;
+
+            get_result_by_id_from_original_flow(
+                &db,
+                w_id.as_str(),
+                &restarted_from.flow_job_id,
+                node_id.as_str(),
+                json_path.clone(),
+            )
+            .await
+        }
+    }
+}
+
+#[async_recursion]
+async fn get_result_by_id_from_running_flow(
+    db: &Pool<Postgres>,
+    w_id: &str,
+    flow_id: &Uuid,
+    node_id: &str,
+    json_path: Option<String>,
 ) -> error::Result<Box<RawValue>> {
     let flow_job_result = sqlx::query!(
         "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3",
@@ -1481,7 +1530,7 @@ pub async fn get_result_by_id(
         flow_id,
         w_id,
     )
-    .fetch_optional(&db)
+    .fetch_optional(db)
     .await?;
 
     let flow_job_result = windmill_common::utils::not_found_if_none(
@@ -1498,11 +1547,11 @@ pub async fn get_result_by_id(
     if job_result.is_none() && flow_job_result.parent_job.is_some() {
         let parent_job = flow_job_result.parent_job.unwrap();
         let root_job = sqlx::query_scalar!("SELECT root_job FROM queue WHERE id = $1", parent_job)
-            .fetch_optional(&db)
+            .fetch_optional(db)
             .await?
             .flatten()
             .unwrap_or(parent_job);
-        return get_result_by_id(db, w_id, root_job, node_id, json_path).await;
+        return get_result_by_id_from_running_flow(db, w_id, &root_job, node_id, json_path).await;
     }
 
     let result_id = windmill_common::utils::not_found_if_none(
@@ -1511,21 +1560,181 @@ pub async fn get_result_by_id(
         format!("{}, {}", flow_id, node_id),
     )?;
 
-    let value = match result_id {
-        JobResult::ListJob(x) => {
+    extract_result_from_job_result(db, w_id, result_id, json_path).await
+}
+
+#[async_recursion]
+async fn get_result_by_id_from_original_flow(
+    db: &Pool<Postgres>,
+    w_id: &str,
+    completed_flow_id: &Uuid,
+    node_id: &str,
+    json_path: Option<String>,
+) -> error::Result<Box<RawValue>> {
+    let flow_job = sqlx::query_as::<_, CompletedJob>(
+        "SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
+    )
+    .bind(completed_flow_id)
+    .bind(w_id)
+    .fetch_optional(db)
+    .await?;
+
+    let flow_job = windmill_common::utils::not_found_if_none(
+        flow_job,
+        "Flow result by id in leaf jobs",
+        format!("{}", completed_flow_id),
+    )?;
+
+    let mut leaf_jobs_for_flow = HashMap::<String, JobResult>::new();
+    compute_leaf_jobs_for_completed_flow(
+        db,
+        w_id,
+        flow_job.id,
+        Some(flow_job.clone()),
+        &mut leaf_jobs_for_flow,
+    )
+    .await?;
+
+    tracing::debug!(
+        "Fetching leaf jobs for flow {} : {:?}",
+        flow_job.id,
+        leaf_jobs_for_flow
+    );
+    if !leaf_jobs_for_flow.contains_key(&node_id.to_string()) {
+        // if the flow is itself a restart flow, the step job might be from the upstream flow
+        let restarted_from = windmill_common::utils::not_found_if_none(
+            flow_job
+                .parse_flow_status()
+                .map(|status| status.restarted_from)
+                .flatten(),
+            "Flow result by id in leaf jobs",
+            format!("{}", completed_flow_id),
+        )?;
+        return get_result_by_id_from_original_flow(
+            db,
+            w_id,
+            &restarted_from.flow_job_id,
+            node_id,
+            json_path,
+        )
+        .await;
+    }
+
+    // if the job is in the leaf_jobs map, then fetch its result and return
+    let leaf_job_uuid = leaf_jobs_for_flow.get(&node_id.to_string()).unwrap();
+    extract_result_from_job_result(db, w_id, leaf_job_uuid.to_owned(), json_path).await
+}
+
+#[async_recursion]
+async fn compute_leaf_jobs_for_completed_flow(
+    db: &Pool<Postgres>,
+    w_id: &str,
+    completed_flow_id: Uuid,
+    completed_flow_row: Option<CompletedJob>, // if provided, will be used, otherwise completed_flow_id must be set and the job definition will be pulled from DB
+    recursive_result: &mut HashMap<String, JobResult>,
+) -> error::Result<()> {
+    let flow_status = match completed_flow_row {
+        Some(job) => job.parse_flow_status(),
+        None => {
+            let job_status_raw = sqlx::query_scalar!(
+                "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
+                completed_flow_id,
+                w_id,
+            )
+            .fetch_one(db)
+            .await?;
+            job_status_raw
+                .map(|raw| serde_json::from_value::<FlowStatus>(raw).ok())
+                .flatten()
+        }
+    };
+
+    let flow_job_status_modules = windmill_common::utils::not_found_if_none(
+        flow_status.map(|fs| fs.modules),
+        "Flow result by id in leaf jobs",
+        format!("{}", completed_flow_id),
+    )?;
+
+    let children_jobs = sqlx::query_as::<_, (Uuid, JobKind)>(
+        "SELECT id, job_kind FROM completed_job WHERE parent_job = $1 AND workspace_id = $2",
+    )
+    .bind(completed_flow_id)
+    .bind(w_id)
+    .fetch_all(db)
+    .await?;
+
+    for child_job in children_jobs {
+        let child_job_id = child_job.0;
+        let child_job_kind = child_job.1;
+        match child_job_kind {
+            JobKind::Script | JobKind::Preview | JobKind::Script_Hub => {
+                // if is potentially a leaf job. Get its step_id from the initial flow definition and add it to the result map
+                for module in &flow_job_status_modules {
+                    if module.job().map(|id| id == child_job_id).unwrap_or(false) {
+                        recursive_result.insert(module.id(), JobResult::SingleJob(child_job_id));
+                    }
+                }
+            }
+            JobKind::Flow | JobKind::FlowPreview => {
+                // Extract the leaf job for this flow and add them to the result map
+                for module in &flow_job_status_modules {
+                    // we add the module as an element of ListJob for this step ID and recursiively extract leaf job of the sub-flow
+                    match module {
+                        FlowStatusModule::Success { flow_jobs: Some(jobs), .. } => {
+                            for job in jobs {
+                                if *job == child_job_id {
+                                    let new_list_job = match recursive_result.get(&module.id()) {
+                                        Some(JobResult::ListJob(jobs_list)) => jobs_list
+                                            .into_iter()
+                                            .chain(iter::once(&child_job_id))
+                                            .cloned()
+                                            .collect_vec(),
+                                        _ => iter::once(&child_job_id).cloned().collect_vec(),
+                                    };
+                                    recursive_result
+                                        .insert(module.id(), JobResult::ListJob(new_list_job));
+                                }
+                            }
+                        }
+                        _ => {}
+                    }
+                }
+                compute_leaf_jobs_for_completed_flow(
+                    db,
+                    w_id,
+                    child_job_id,
+                    None,
+                    recursive_result,
+                )
+                .await?;
+            }
+            _ => {} // do nothing
+        }
+    }
+    Ok(())
+}
+
+async fn extract_result_from_job_result(
+    db: &Pool<Postgres>,
+    w_id: &str,
+    job_result: JobResult,
+    json_path: Option<String>,
+) -> error::Result<Box<RawValue>> {
+    match job_result {
+        JobResult::ListJob(job_ids) => {
             let rows = sqlx::query(
                 "SELECT result FROM completed_job WHERE id = ANY($1) AND workspace_id = $2",
             )
-            .bind(x.as_slice())
+            .bind(job_ids.as_slice())
             .bind(w_id)
-            .fetch_all(&db)
+            .fetch_all(db)
             .await?
             .into_iter()
             .filter_map(|x| ResultR::from_row(&x).ok().and_then(|x| x.result))
             .collect::<Vec<Json<Box<RawValue>>>>();
-            to_raw_value(&rows)
+            Ok(to_raw_value(&rows))
         }
-        JobResult::SingleJob(x) => sqlx::query(
+        JobResult::SingleJob(x) => Ok(sqlx::query(
             "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
         )
         .bind(x)
@@ -1535,7 +1744,7 @@ pub async fn get_result_by_id(
                 .map(|x| x.split(".").map(|x| x.to_string()).collect::<Vec<_>>())
                 .unwrap_or_default(),
         )
-        .fetch_optional(&db)
+        .fetch_optional(db)
         .await?
         .map(|r| {
             ResultR::from_row(&r)
@@ -1543,10 +1752,8 @@ pub async fn get_result_by_id(
                 .and_then(|x| x.result.map(|x| x.0))
         })
         .flatten()
-        .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)),
-    };
-
-    Ok(value)
+        .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))),
+    }
 }
 
 #[instrument(level = "trace", skip_all)]
@@ -1896,6 +2103,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
         raw_code_tuple,
         job_kind,
         mut raw_flow,
+        flow_status,
         language,
         concurrent_limit,
         concurrency_time_window_s,
@@ -1918,6 +2126,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             None,
             JobKind::Script,
             None,
+            None,
             Some(language),
             concurrent_limit,
             concurrency_time_window_s,
@@ -1939,6 +2148,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                 None,
                 None,
                 None,
+                None,
             )
         }
         JobPayload::Code(RawCode {
@@ -1955,6 +2165,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             Some((content, lock)),
             JobKind::Preview,
             None,
+            None,
             Some(language),
             concurrent_limit,
             concurrency_time_window_s,
@@ -1968,6 +2179,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             Some((dependencies, None)),
             JobKind::Dependencies,
             None,
+            None,
             Some(language),
             None,
             None,
@@ -1995,7 +2207,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                 Some(path),
                 None,
                 JobKind::FlowDependencies,
-                Some(value),
+                Some(value.clone()),
+                Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value
                 None,
                 None,
                 None,
@@ -2016,20 +2229,56 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             None,
             None,
             None,
+            None,
         ),
-        JobPayload::RawFlow { value, path } => (
-            None,
-            path,
-            None,
-            JobKind::FlowPreview,
-            Some(value.clone()),
-            None,
-            value.concurrent_limit.clone(),
-            value.concurrency_time_window_s,
-            value.cache_ttl.map(|x| x as i32),
-            None,
-            value.priority,
-        ),
+        JobPayload::RawFlow { value, path, restarted_from } => {
+            let flow_status: FlowStatus = match restarted_from {
+                Some(restarted_from_val) => {
+                    let (_, _, step_n, truncated_modules, _) = restarted_flows_resolution(
+                        _db,
+                        workspace_id,
+                        Some(value.clone()),
+                        restarted_from_val.flow_job_id,
+                        restarted_from_val.step_id.as_str(),
+                    )
+                    .await?;
+                    FlowStatus {
+                        step: step_n,
+                        modules: truncated_modules,
+                        // failure_module is reset
+                        failure_module: FlowStatusModuleWParent {
+                            parent_module: None,
+                            module_status: FlowStatusModule::WaitingForPriorSteps {
+                                id: "failure".to_string(),
+                            },
+                        },
+                        // retry status is reset
+                        retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
+                        // TODO: for now, flows with approval conditions aren't supported for restart
+                        approval_conditions: None,
+                        restarted_from: Some(RestartedFrom {
+                            flow_job_id: restarted_from_val.flow_job_id,
+                            step_id: restarted_from_val.step_id,
+                        }),
+                    }
+                }
+                _ => FlowStatus::new(&value), // this is a new flow being pushed, flow_status is set to flow_value
+            };
+            (
+                None,
+                path,
+                None,
+                JobKind::FlowPreview,
+                Some(value.clone()),
+                Some(flow_status),
+                None,
+                value.concurrent_limit.clone(),
+                value.concurrency_time_window_s,
+                value.cache_ttl.map(|x| x as i32),
+                None,
+                value.priority,
+            )
+        }
         JobPayload::Flow(flow) => {
             let value_json = fetch_scalar_isolated!(
                 sqlx::query_scalar!(
@@ -2051,6 +2300,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                 None,
                 JobKind::Flow,
                 Some(value.clone()),
+                Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value
                 None,
                 value.concurrent_limit.clone(),
                 value.concurrency_time_window_s,
@@ -2059,6 +2309,47 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                 value.priority,
             )
         }
+        JobPayload::RestartedFlow { completed_job_id, step_id } => {
+            let (flow_path, raw_flow, step_n, truncated_modules, priority) =
+                restarted_flows_resolution(
+                    _db,
+                    workspace_id,
+                    None,
+                    completed_job_id,
+                    step_id.as_str(),
+                )
+                .await?;
+            let restarted_flow_status = FlowStatus {
+                step: step_n,
+                modules: truncated_modules,
+                // failure_module is reset
+                failure_module: FlowStatusModuleWParent {
+                    parent_module: None,
+                    module_status: FlowStatusModule::WaitingForPriorSteps {
+                        id: "failure".to_string(),
+                    },
+                },
+                // retry status is reset
+                retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
+                // TODO: for now, flows with approval conditions aren't supported for restart
+                approval_conditions: None,
+                restarted_from: Some(RestartedFrom { flow_job_id: completed_job_id, step_id }),
+            };
+            (
+                None,
+                flow_path,
+                None,
+                JobKind::Flow,
+                Some(raw_flow.clone()),
+                Some(restarted_flow_status),
+                None,
+                raw_flow.concurrent_limit,
+                raw_flow.concurrency_time_window_s,
+                raw_flow.cache_ttl.map(|x| x as i32),
+                None,
+                priority,
+            )
+        }
         JobPayload::Identity => (
             None,
             None,
@@ -2071,6 +2362,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             None,
             None,
             None,
+            None,
         ),
         JobPayload::Noop => (
             None,
@@ -2084,6 +2376,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             None,
             None,
             None,
+            None,
         ),
     };
 
@@ -2156,8 +2449,6 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
         .map(|e| (Some(e.0), e.1))
         .unwrap_or_else(|| (None, None));
 
-    let flow_status = raw_flow.as_ref().map(FlowStatus::new);
-
     let tag = if dedicated_worker.is_some_and(|x| x) {
         format!(
             "{}:{}",
@@ -2317,3 +2608,102 @@ pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
     let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
     serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
 }
+
+async fn restarted_flows_resolution(
+    db: &Pool<Postgres>,
+    workspace_id: &str,
+    flow_value_if_any: Option<FlowValue>,
+    completed_flow_id: Uuid,
+    restart_step_id: &str,
+) -> Result<
+    (
+        Option<String>,
+        FlowValue,
+        i32,
+        Vec<FlowStatusModule>,
+        Option<i16>,
+    ),
+    Error,
+> {
+    let completed_job = sqlx::query_as::<_, CompletedJob>(
+        "SELECT * FROM completed_job WHERE id = $1 and workspace_id = $2",
+    )
+    .bind(completed_flow_id)
+    .bind(workspace_id)
+    .fetch_one(db) // TODO: should we try to use the passed-in `tx` here?
+    .await
+    .map_err(|err| {
+        Error::InternalErr(format!(
+            "completed job not found for UUID {} in workspace {}: {}",
+            completed_flow_id, workspace_id, err
+        ))
+    })?;
+
+    let raw_flow = completed_job
+        .parse_raw_flow()
+        .ok_or(Error::InternalErr(format!(
+            "Unable to parse raw definition for job {} in workspace {}",
+            completed_flow_id, workspace_id,
+        )))?;
+    let flow_status = completed_job
+        .parse_flow_status()
+        .ok_or(Error::InternalErr(format!(
+            "Unable to parse flow status for job {} in workspace {}",
+            completed_flow_id, workspace_id,
+        )))?;
+
+    let mut step_n = 0;
+    let mut dependent_module = false;
+    let mut truncated_modules: Vec<FlowStatusModule> = vec![];
+    for module in flow_status.modules {
+        if flow_value_if_any
+            .clone()
+            .map(|fv| {
+                fv.modules
+                    .iter()
+                    .find(|flow_value_module| flow_value_module.id == module.id())
+                    .is_none()
+            })
+            .unwrap_or(false)
+        {
+            // skip module as it doesn't appear in the flow_value anymore
+            continue;
+        }
+        if module.id() == restart_step_id || dependent_module {
+            // if the module ID is the one we want to restart the flow at, or if it's past it in the flow,
+            // set the module as WaitingForPriorSteps as it needs to be re-run
+            truncated_modules.push(FlowStatusModule::WaitingForPriorSteps { id: module.id() });
+            dependent_module = true;
+        } else {
+            // else we simply "transfer" the module from the completed flow to the new one if it's a success
+            step_n = step_n + 1;
+            match module.clone() {
+                FlowStatusModule::Success {
+                    id: _,
+                    job: _,
+                    flow_jobs: _,
+                    branch_chosen: _,
+                    approvers: _,
+                } => Ok(truncated_modules.push(module)),
+                _ => Err(Error::InternalErr(format!(
+                    "Flow cannot be restarted from a non successful module",
+                ))),
+            }?;
+        }
+    }
+    if !dependent_module {
+        // step not found in flow.
+        return Err(Error::InternalErr(format!(
+            "Flow cannot be restarted from step {} as it could not be found.",
+            restart_step_id
+        )));
+    }
+
+    return Ok((
+        completed_job.script_path,
+        raw_flow,
+        step_n,
+        truncated_modules,
+        completed_job.priority,
+    ));
+}

backend/windmill-worker/src/worker.rs

@@ -2150,13 +2150,12 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
     };
     match job.job_kind {
         JobKind::FlowPreview | JobKind::Flow => {
-            let args = job.get_args();
             let timer = worker_flow_initial_transition_duration.map(|x| x.start_timer());
             handle_flow(
                 &job,
                 db,
                 &client.get_authed().await,
-                to_raw_value(&args),
+                None,
                 same_worker_tx,
                 worker_dir,
                 rsmq,

backend/windmill-worker/src/worker_flow.rs

@@ -28,7 +28,8 @@ use windmill_common::flow_status::{
     ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
 };
 use windmill_common::jobs::{
-    script_hash_to_tag_and_limits, script_path_to_payload, JobPayload, QueuedJob, RawCode,
+    script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob,
+    RawCode,
 };
 use windmill_common::worker::to_raw_value;
 use windmill_common::{
@@ -137,12 +138,6 @@ pub struct RecUpdateFlowStatusAfterJobCompletion {
     skip_error_handler: bool,
 }
 
-#[derive(FromRow)]
-pub struct BranchResults<'a> {
-    pub result: &'a RawValue,
-    pub id: Uuid,
-}
-
 #[derive(FromRow)]
 pub struct SkipIfStopped {
     pub skip_if_stopped: Option<bool>,
@@ -508,35 +503,7 @@ pub async fn update_flow_status_after_job_completion_internal<
         let nresult = match &new_status {
             Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. })
             | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => {
-                let results = sqlx::query(
-                    "
-                  SELECT result, id
-                    FROM completed_job
-                   WHERE id = ANY($1)
-                     AND workspace_id = $2
-                    ",
-                )
-                .bind(jobs.as_slice())
-                .bind(w_id)
-                .fetch_all(&mut tx)
-                .await?
-                .into_iter()
-                .map(|r| {
-                    let br = BranchResults::from_row(&r).unwrap();
-                    (br.id, br.result.to_owned())
-                })
-                .collect::<HashMap<_, _>>();
-
-                let results = jobs
-                    .iter()
-                    .map(|j| {
-                        results.get(j).ok_or_else(|| {
-                            Error::InternalErr(format!("missing job result for {}", j))
-                        })
-                    })
-                    .collect::<Result<Vec<_>, _>>()?;
-
-                to_raw_value(&results)
+                retrieve_flow_jobs_results(db, w_id, jobs).await?
             }
             _ => result.to_owned(),
         };
@@ -690,7 +657,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             &flow_job,
             db,
             client,
-            nresult.to_owned(),
+            Some(nresult.to_owned()),
             same_worker_tx.clone(),
             worker_dir,
             rsmq.clone(),
@@ -741,6 +708,42 @@ pub async fn update_flow_status_after_job_completion_internal<
     }
 }
 
+async fn retrieve_flow_jobs_results(
+    db: &DB,
+    w_id: &str,
+    job_uuids: &Vec<Uuid>,
+) -> error::Result<Box<RawValue>> {
+    let results = sqlx::query(
+        "
+      SELECT result, id
+        FROM completed_job
+       WHERE id = ANY($1)
+         AND workspace_id = $2
+        ",
+    )
+    .bind(job_uuids.as_slice())
+    .bind(w_id)
+    .fetch_all(db)
+    .await?
+    .into_iter()
+    .map(|r| {
+        let br = BranchResults::from_row(&r).unwrap();
+        (br.id, br.result.to_owned())
+    })
+    .collect::<HashMap<_, _>>();
+
+    let results = job_uuids
+        .iter()
+        .map(|j| {
+            results
+                .get(j)
+                .ok_or_else(|| Error::InternalErr(format!("missing job result for {}", j)))
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
+    Ok(to_raw_value(&results))
+}
+
 fn get_module(flow_job: &QueuedJob, module_index: Option<usize>) -> Option<FlowModule> {
     let raw_flow = flow_job.parse_raw_flow();
     if let Some(raw_flow) = raw_flow {
@@ -835,7 +838,7 @@ async fn compute_bool_from_expr(
 ) -> error::Result<bool> {
     let mut context = HashMap::with_capacity(if resumes.is_some() { 7 } else { 3 });
     context.insert("result".to_string(), result.clone());
-    context.insert("previous_result".to_string(), result);
+    context.insert("previous_result".to_string(), result.clone());
 
     if let Some(resumes) = resumes {
         context.insert("resume".to_string(), resumes.1);
@@ -984,25 +987,19 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
     flow_job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClient,
-    last_result: Box<RawValue>,
+    last_result: Option<Box<RawValue>>,
     same_worker_tx: Sender<Uuid>,
     worker_dir: &str,
     rsmq: Option<R>,
     worker_name: &str,
 ) -> 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_str::<FlowValue>((*value.0).get())?;
+    let flow = flow_job
+        .parse_raw_flow()
+        .with_context(|| "Unable to parse flow definition")?;
+    let status = flow_job
+        .parse_flow_status()
+        .with_context(|| "Unable to parse flow status")?;
 
-    let status: FlowStatus = serde_json::from_str::<FlowStatus>(
-        (*flow_job.flow_status.clone().unwrap_or_default().0).get(),
-    )
-    .with_context(|| format!("parse flow status {}", flow_job.id))?;
-
-    tracing::debug!("handle_flow: {:#?}", flow_job.flow_status);
     push_next_flow_job(
         flow_job,
         status,
@@ -1054,7 +1051,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     flow: FlowValue,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClient,
-    last_result: Box<RawValue>,
+    last_job_result: Option<Box<RawValue>>,
     same_worker_tx: Sender<Uuid>,
     worker_dir: &str,
     rsmq: Option<R>,
@@ -1105,14 +1102,13 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     }
 
     let arc_flow_job_args = Arc::new(flow_job_args.clone());
-    let mut arc_result = Arc::new(last_result);
 
     if i == 0 {
         if let Some(skip_expr) = &flow.skip_expr {
             let skip = compute_bool_from_expr(
                 skip_expr.to_string(),
                 arc_flow_job_args.clone(),
-                arc_result.clone(),
+                Arc::new(to_raw_value(&json!("{}"))),
                 None,
                 Some(client),
                 None,
@@ -1139,6 +1135,27 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         }
     }
 
+    // Compute and initialize last_job_result
+    let arc_last_job_result = if status_module.is_failure() {
+        // if job is being retried, pass the result of its previous failure
+        Arc::new(last_job_result.unwrap_or(to_raw_value(&json!("{}"))))
+    } else if i == 0 {
+        // if it's the first job executed in the flow, pass the flow args
+        Arc::new(to_raw_value(&flow_job.args))
+    } else {
+        // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job
+        // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a
+        // flow restart from a specific step
+        if last_job_result.is_some() {
+            Arc::new(last_job_result.unwrap())
+        } else {
+            match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? {
+                None => Arc::new(to_raw_value(&json!("{}"))),
+                Some(previous_job_result) => Arc::new(previous_job_result),
+            }
+        }
+    };
+
     let mut resume_messages: Vec<Box<RawValue>> = vec![];
     let mut approvers: Vec<String> = vec![];
 
@@ -1202,8 +1219,9 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                         }
                         InputTransform::Javascript { expr } => {
                             let mut context = HashMap::with_capacity(2);
-                            context.insert("result".to_string(), arc_result.clone());
-                            context.insert("previous_result".to_string(), arc_result.clone());
+                            context.insert("result".to_string(), arc_last_job_result.clone());
+                            context
+                                .insert("previous_result".to_string(), arc_last_job_result.clone());
 
                             let eval_result = serde_json::from_str::<Vec<String>>(
                                 eval_timeout(
@@ -1275,20 +1293,6 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 .execute(&mut *tx)
                 .await?;
 
-                /* If we are woken up after suspending, last_result will be the flow args, but we
-                 * should use the result from the last job */
-                if let FlowStatusModule::WaitingForEvents { .. } = &status_module {
-                    arc_result = Arc::new(
-                        sqlx::query_scalar::<_, Json<Box<RawValue>>>(
-                            "SELECT result FROM completed_job WHERE id = $1",
-                        )
-                        .bind(last)
-                        .fetch_one(&mut *tx)
-                        .await?
-                        .0,
-                    )
-                }
-
                 // Remove the approval conditions from the flow status
                 sqlx::query(
                     "
@@ -1386,8 +1390,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     InputTransform::Static { value } => Ok(value),
                     InputTransform::Javascript { expr } => {
                         let mut context = HashMap::with_capacity(2);
-                        context.insert("result".to_string(), arc_result.clone());
-                        context.insert("previous_result".to_string(), arc_result.clone());
+                        context.insert("result".to_string(), arc_last_job_result.clone());
+                        context.insert("previous_result".to_string(), arc_last_job_result.clone());
 
                         serde_json::from_str(
                             eval_timeout(
@@ -1562,7 +1566,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 let by_id = transform_context.as_ref().unwrap();
                 transform_input(
                     arc_flow_job_args.clone(),
-                    arc_result.clone(),
+                    arc_last_job_result.clone(),
                     input_transforms,
                     resumes.clone(),
                     resume.clone(),
@@ -1573,8 +1577,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 .await
             }
             FlowModuleValue::Identity => serde_json::from_str(
-                &serde_json::to_string(&PreviousResult { previous_result: Some(&arc_result) })
-                    .unwrap(),
+                &serde_json::to_string(&PreviousResult {
+                    previous_result: Some(&arc_last_job_result),
+                })
+                .unwrap(),
             )
             .map_err(|e| error::Error::InternalErr(format!("identity: {e}"))),
 
@@ -1584,7 +1590,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
     let next_flow_transform = compute_next_flow_transform(
         arc_flow_job_args.clone(),
-        arc_result.clone(),
+        arc_last_job_result.clone(),
         flow_job,
         &flow,
         transform_context,
@@ -1701,7 +1707,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                         let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
                         transform_inp = transform_input(
                             Arc::new(hm),
-                            arc_result.clone(),
+                            arc_last_job_result.clone(),
                             input_transforms,
                             resumes.clone(),
                             resume.clone(),
@@ -2013,7 +2019,7 @@ enum NextFlowTransform {
 
 async fn compute_next_flow_transform(
     arc_flow_job_args: Arc<HashMap<String, Box<RawValue>>>,
-    arc_result: Arc<Box<RawValue>>,
+    arc_last_job_result: Arc<Box<RawValue>>,
     flow_job: &QueuedJob,
     flow: &FlowValue,
     by_id: Option<IdContext>,
@@ -2106,8 +2112,8 @@ async fn compute_next_flow_transform(
                         InputTransform::Static { value } => to_raw_value(value),
                         InputTransform::Javascript { expr } => {
                             let mut context = HashMap::with_capacity(3);
-                            context.insert("result".to_string(), arc_result.clone());
-                            context.insert("previous_result".to_string(), arc_result);
+                            context.insert("result".to_string(), arc_last_job_result.clone());
+                            context.insert("previous_result".to_string(), arc_last_job_result);
                             context.insert("resumes".to_string(), resumes);
                             context.insert("resume".to_string(), resume);
                             context.insert("approvers".to_string(), approvers);
@@ -2197,6 +2203,7 @@ async fn compute_next_flow_transform(
                                 priority: None,
                             },
                             path: inner_path,
+                            restarted_from: None,
                         },
                         tag: None,
                     });
@@ -2256,6 +2263,7 @@ async fn compute_next_flow_transform(
                                         priority: None,
                                     },
                                     path: Some(format!("{}/forloop", flow_job.script_path())),
+                                    restarted_from: None,
                                 },
                                 tag: None,
                             }
@@ -2298,7 +2306,7 @@ async fn compute_next_flow_transform(
                         let pred = compute_bool_from_expr(
                             b.expr.to_string(),
                             arc_flow_job_args.clone(),
-                            arc_result.clone(),
+                            arc_last_job_result.clone(),
                             Some(idcontext.clone()),
                             Some(client),
                             Some((resumes.clone(), resume.clone(), approvers.clone())),
@@ -2353,6 +2361,7 @@ async fn compute_next_flow_transform(
                             flow_job.script_path(),
                             status.step
                         )),
+                        restarted_from: None,
                     },
                     tag: None,
                 }),
@@ -2399,6 +2408,7 @@ async fn compute_next_flow_transform(
                                                     flow_job.script_path(),
                                                     i
                                                 )),
+                                                restarted_from: None,
                                             },
                                             tag: None,
                                         }
@@ -2462,6 +2472,7 @@ async fn compute_next_flow_transform(
                             flow_job.script_path(),
                             branch_status.branch
                         )),
+                        restarted_from: None,
                     },
                     tag: None,
                 }),
@@ -2574,8 +2585,15 @@ fn from_now(duration: Duration) -> chrono::DateTime<chrono::Utc> {
         .unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
 }
 
-/// returns previous module non-zero suspend count and job
+/// returns previous module non-zero suspend count and job, if relevant
 fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid)> {
+    // for a restarted job, if the restarted step is just after a suspend, don't run the suspend
+    if status.restarted_from.is_some() {
+        let current_step_id = flow.modules.get(status.step as usize)?.id.clone();
+        if status.restarted_from.as_ref().unwrap().step_id == current_step_id {
+            return None;
+        }
+    }
     let prev = usize::try_from(status.step)
         .ok()
         .and_then(|s| s.checked_sub(1))?;
@@ -2596,3 +2614,32 @@ fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid)
         None
     }
 }
+
+// returns the result of the previous step of a running flow (if the job was successful)
+async fn get_previous_job_result(
+    db: &sqlx::Pool<sqlx::Postgres>,
+    w_id: &str,
+    flow_status: &FlowStatus,
+) -> error::Result<Option<Box<RawValue>>> {
+    let prev = usize::try_from(flow_status.step)
+        .ok()
+        .and_then(|s| s.checked_sub(1))
+        .with_context(|| "No step preceding the current one")?;
+
+    match flow_status.modules.get(prev) {
+        Some(FlowStatusModule::Success { flow_jobs: Some(flow_jobs), .. }) => {
+            Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?))
+        }
+        Some(FlowStatusModule::Success { job, .. }) => Ok(Some(
+            sqlx::query_scalar::<_, Json<Box<RawValue>>>(
+                "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
+            )
+            .bind(job)
+            .bind(w_id)
+            .fetch_one(db)
+            .await?
+            .0,
+        )),
+        _ => Ok(None),
+    }
+}

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -1,8 +1,8 @@
 <script lang="ts">
-	import { Job, JobService, type Flow, type FlowModule } from '$lib/gen'
+	import { Job, JobService, type Flow, type FlowModule, type RestartedFrom } from '$lib/gen'
 	import { workspaceStore } from '$lib/stores'
 	import { faClose, faPlay, faRefresh } from '@fortawesome/free-solid-svg-icons'
-	import { Button, Drawer, Kbd } from './common'
+	import { Badge, Button, Drawer, Kbd } from './common'
 	import { createEventDispatcher, getContext } from 'svelte'
 	import Icon from 'svelte-awesome'
 	import type { FlowEditorContext } from './flows/types'
@@ -23,6 +23,7 @@
 
 	export let jobId: string | undefined = undefined
 	export let job: Job | undefined = undefined
+	let selectedJobStep: string | undefined = undefined
 	let isRunning: boolean = false
 	let jobProgressReset: () => void
 
@@ -70,10 +71,13 @@
 		}
 	}
 
-	export async function runPreview(args: Record<string, any>) {
+	export async function runPreview(
+		args: Record<string, any>,
+		restartedFrom: RestartedFrom | undefined
+	) {
 		jobProgressReset()
 		const newFlow = extractFlow(previewMode)
-		jobId = await runFlowPreview(args, newFlow)
+		jobId = await runFlowPreview(args, newFlow, restartedFrom)
 		isRunning = true
 	}
 
@@ -83,7 +87,7 @@
 				case 'Enter':
 					if (event.ctrlKey || event.metaKey) {
 						event.preventDefault()
-						runPreview($previewArgs)
+						runPreview($previewArgs, undefined)
 					}
 					break
 			}
@@ -150,18 +154,42 @@
 				Cancel
 			</Button>
 		{:else}
-			<Button
-				variant="contained"
-				startIcon={{ icon: isRunning ? faRefresh : faPlay }}
-				color="dark"
-				size="sm"
-				btnClasses="w-full max-w-lg"
-				on:click={() => runPreview($previewArgs)}
-				id="flow-editor-test-flow-drawer"
-			>
-				Test flow &nbsp;<Kbd small isModifier>{getModifierKey()}</Kbd>
-				<Kbd small><span class="text-lg font-bold">⏎</span></Kbd>
-			</Button>
+			<div class="flex flex-row gap-4">
+				{#if jobId !== undefined && selectedJobStep !== undefined && job?.flow_status?.modules !== undefined && job?.flow_status?.modules
+						.map((m) => m.id)
+						.indexOf(selectedJobStep) >= 0}
+					<Button
+						size="xs"
+						color="light"
+						variant="border"
+						title={`Re-start this flow from step ${selectedJobStep} (included).`}
+						on:click={() => {
+							runPreview($previewArgs, {
+								flow_job_id: jobId,
+								step_id: selectedJobStep
+							})
+						}}
+						startIcon={{ icon: faPlay }}
+					>
+						Re-start from
+						<Badge baseClass="ml-1" color="indigo">
+							{selectedJobStep}
+						</Badge>
+					</Button>
+				{/if}
+				<Button
+					variant="contained"
+					startIcon={{ icon: isRunning ? faRefresh : faPlay }}
+					color="dark"
+					size="sm"
+					btnClasses="w-full max-w-lg"
+					on:click={() => runPreview($previewArgs, undefined)}
+					id="flow-editor-test-flow-drawer"
+				>
+					Test flow &nbsp;<Kbd small isModifier>{getModifierKey()}</Kbd>
+					<Kbd small><span class="text-lg font-bold">⏎</span></Kbd>
+				</Button>
+			</div>
 		{/if}
 		<div class="flex gap-2">
 			{#if initialPath != ''}
@@ -205,6 +233,7 @@
 					on:jobsLoaded={({ detail }) => {
 						job = detail
 					}}
+					bind:selectedJobStep
 				/>
 			{:else}
 				<div class="italic text-tertiary h-full grow"> Flow status will be displayed here </div>

frontend/src/lib/components/FlowStatusViewer.svelte

@@ -10,6 +10,7 @@
 	export let jobId: string
 	export let workspaceId: string | undefined = undefined
 	export let flowStateStore: Writable<FlowState> | undefined = undefined
+	export let selectedJobStep: string | undefined = undefined
 
 	export let isOwner = false
 
@@ -56,6 +57,7 @@
 		}
 		dispatch('jobsLoaded', detail)
 	}}
+	bind:selectedNode={selectedJobStep}
 	{jobId}
 	{workspaceId}
 	{isOwner}

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -39,7 +39,7 @@
 
 	export let isOwner = false
 
-	let selectedNode: string | undefined = undefined
+	export let selectedNode: string | undefined = undefined
 
 	let jobResults: any[] = []
 	let jobFailures: boolean[] = []

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

@@ -4,7 +4,8 @@ import {
 	type Flow,
 	type FlowModule,
 	type InputTransform,
-	type Job
+	type Job,
+	type RestartedFrom
 } from '$lib/gen'
 import { workspaceStore } from '$lib/stores'
 import { cleanExpr, emptySchema } from '$lib/utils'
@@ -108,7 +109,7 @@ export function jobsToResults(jobs: Job[]) {
 	})
 }
 
-export async function runFlowPreview(args: Record<string, any>, flow: Flow) {
+export async function runFlowPreview(args: Record<string, any>, flow: Flow, restartedFrom: RestartedFrom | undefined) {
 	const newFlow = flow
 	return await JobService.runFlowPreview({
 		workspace: get(workspaceStore) ?? '',
@@ -116,7 +117,8 @@ export async function runFlowPreview(args: Record<string, any>, flow: Flow) {
 			args,
 			value: newFlow.value,
 			path: newFlow.path,
-			tag: newFlow.tag
+			tag: newFlow.tag,
+			restarted_from: restartedFrom,
 		}
 	})
 }

frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte

@@ -18,7 +18,14 @@
 		faFastForward
 	} from '@fortawesome/free-solid-svg-icons'
 	import DisplayResult from '$lib/components/DisplayResult.svelte'
-	import { runFormStore, superadmin, userStore, userWorkspaces, workspaceStore } from '$lib/stores'
+	import {
+		enterpriseLicense,
+		runFormStore,
+		superadmin,
+		userStore,
+		userWorkspaces,
+		workspaceStore
+	} from '$lib/stores'
 	import CenteredPage from '$lib/components/CenteredPage.svelte'
 	import FlowStatusViewer from '$lib/components/FlowStatusViewer.svelte'
 	import HighlightCode from '$lib/components/HighlightCode.svelte'
@@ -40,6 +47,7 @@
 	const iconScale = 1
 
 	let viewTab: 'result' | 'logs' | 'code' = 'result'
+	let selectedJobStep: string | undefined = undefined
 
 	// Test
 	let testIsLoading = false
@@ -67,6 +75,19 @@
 		}
 	}
 
+	async function restartFlow(id: string | undefined, stepId: string | undefined) {
+		if (id === undefined || stepId === undefined) {
+			return
+		}
+		let run = await JobService.restartFlowAtStep({
+			workspace: $workspaceStore!,
+			id,
+			stepId,
+			requestBody: {}
+		})
+		await goto('/run/' + run + '?workspace=' + $workspaceStore)
+	}
+
 	// If we get results, focus on that tab. Else, focus on logs
 	function initView(): void {
 		if (job && 'result' in job && job.result != undefined) {
@@ -199,6 +220,27 @@
 					</Button>
 				{/if}
 			{/if}
+			{#if job?.job_kind === 'flow' && selectedJobStep !== undefined && job?.flow_status?.modules !== undefined && job?.flow_status?.modules
+					.map((m) => m.id)
+					.indexOf(selectedJobStep) >= 0}
+				<Button
+					title={`Re-start this flow from step ${selectedJobStep} (included). ${
+						!$enterpriseLicense ? ' This is a feature only available in enterprise edition.' : ''
+					}`}
+					variant="border"
+					color="blue"
+					disabled={!$enterpriseLicense}
+					on:click|once={() => {
+						restartFlow(job?.id, selectedJobStep)
+					}}
+					startIcon={{ icon: faRefresh }}
+				>
+					Re-start from
+					<Badge baseClass="ml-1" color="indigo">
+						{selectedJobStep}
+					</Badge>
+				</Button>
+			{/if}
 			{#if job?.job_kind === 'script' || job?.job_kind === 'flow'}
 				<Button
 					on:click|once={() => {
@@ -391,6 +433,7 @@
 						job = detail
 					}}
 					workspaceId={$workspaceStore}
+					bind:selectedJobStep
 				/>
 			</div>
 		{/if}