fix(flows): skip_if evaluates wrong previous_result during retry (#9547)

* fix(flows): evaluate skip_if against previous step result on retry

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* docs: narrow retry previous_result restoration comment

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

backend/tests/retry.rs

@@ -262,6 +262,65 @@ def main(last, port):
         Ok(())
     }
 
+    /* `skip_if` evaluation goes through windmill-jseval and requires `quickjs` */
+    #[cfg(all(feature = "deno_core", feature = "quickjs"))]
+    #[sqlx::test(fixtures("base"))]
+    async fn test_skip_if_sees_previous_step_result_on_retry(
+        db: Pool<Postgres>,
+    ) -> anyhow::Result<()> {
+        initialize_tracing().await;
+
+        /* step `b` fails on its first attempt and succeeds on retry. Its `skip_if`
+         * references the previous step's result: when re-evaluated for the retry,
+         * `previous_result` must still be step `a`'s result, not the error of step
+         * `b`'s failed attempt (which would wrongly flip `skip_if` to true and skip
+         * the retry) */
+        let value = serde_json::from_value(json!({
+            "modules": [{
+                "id": "a",
+                "value": {
+                    "input_transforms": {},
+                    "type": "rawscript",
+                    "language": "deno",
+                    "content": "export function main() { return \"ok\" }",
+                },
+            }, {
+                "id": "b",
+                "skip_if": { "expr": "results.a?.error !== undefined" },
+                "value": {
+                    "input_transforms": {
+                        "index": { "type": "static", "value": 1 },
+                        "port": { "type": "javascript", "expr": "flow_input.port" },
+                    },
+                    "type": "rawscript",
+                    "language": "deno",
+                    "content": inner_step(),
+                },
+                "retry": { "constant": { "attempts": 1, "seconds": 0 } },
+            }],
+        }))
+        .unwrap();
+
+        let (attempts, responses) = [
+            /* fail step `b` once, then pass on retry */
+            (1, None),
+            (1, Some(42)),
+        ]
+        .into_iter()
+        .unzip::<_, _, Vec<_>, Vec<_>>();
+        let server = Server::start(responses).await;
+        let result = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None })
+            .arg("port", json!(server.addr.port()))
+            .run_until_complete(&db, false, server.addr.port())
+            .await
+            .json_result()
+            .unwrap();
+
+        assert_eq!(server.close().await, attempts);
+        assert_eq!(json!(42), result);
+        Ok(())
+    }
+
     #[cfg(feature = "python")]
     #[sqlx::test(fixtures("base"))]
     async fn test_with_failure_module(db: Pool<Postgres>) -> anyhow::Result<()> {

backend/windmill-worker/src/worker_flow.rs

@@ -3268,7 +3268,7 @@ async fn push_next_flow_job(
     }
 
     // Compute and initialize last_job_result
-    let arc_last_job_result = if status_module.is_failure() {
+    let mut arc_last_job_result = if status_module.is_failure() {
         // if job is being retried, pass the result of its previous failure
         last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}"))))
     } else if matches!(step, Step::Step { idx: 0, .. }) || step.is_preprocessor_step() {
@@ -3699,6 +3699,30 @@ async fn push_next_flow_job(
                  .context("update flow retry")?;
 
                 status_module = FlowStatusModule::WaitingForPriorSteps { id: status_module.id() };
+
+                // The failed attempt's error has already been consumed by `evaluate_retry`
+                // above. Restore `previous_result` to the preceding step's result (or the
+                // flow args for the first step) so that predicates re-evaluated for the
+                // retry (skip_if, loop iterator expressions, ...) don't see the failed
+                // attempt's error instead. Like the suspend/restart path above, this
+                // falls back to `"{}"` when the preceding step has no Success status
+                // (e.g. it failed with continue_on_error).
+                if !matches!(step, Step::FailureStep) {
+                    arc_last_job_result = if matches!(step, Step::Step { idx: 0, .. })
+                        || step.is_preprocessor_step()
+                    {
+                        Arc::new(to_raw_value(&flow_job.args))
+                    } else {
+                        match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status)
+                            .warn_after_seconds(3)
+                            .await?
+                        {
+                            None => Arc::new(to_raw_value(&json!("{}"))),
+                            Some(previous_job_result) => Arc::new(previous_job_result),
+                        }
+                    };
+                }
+
                 // we get the args from the last failed job
                 status.retry.failed_jobs.last()
             /* Start the failure module ... */