feat(flows): opt-in to include the stopping step's result in early-stop errors (#9446)

* feat(flows): early stop can include the stopping step's result in the raised error

When a step uses Early Stop with "Raise an error message if stopped", the
flow result was entirely replaced with a static error object
({"error": {"name": "EarlyStopError", "message": "..."}}), discarding the
stopping step's own output. This made it impossible to stop+fail a flow
while preserving the data the step produced (e.g. an API that returns
HTTP 200 with a userErrors payload).

Add an opt-in `error_include_result` flag on StopAfterIf. When enabled on
the raise-error path, the raised payload becomes
{"error": {...}, "result": <step result>} instead of dropping the result.
Default is false, so existing behavior is unchanged. The option is threaded
through the worker's stop-after-if handling (including stop_after_all_iters_if
for loops/branchall) and exposed in the flow editor's Early Stop panel.

Fixes WIN-2012

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(flows): cover early-stop error_include_result payload shaping

Add a regression test asserting that a step using Early Stop with a raised
error message and error_include_result=true fails the flow while preserving
the step output as {"error": {..}, "result": <step result>}, and that with
the flag off the result is the bare {"error": {..}} object.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor(flows): nest early-stop step result inside the error object

Embed the stopping step's result under `error.result` rather than as a
top-level sibling of `error`. This keeps the flow result shape as
`{ "error": { .. } }` — identical to a normal error — so consumers that
key off the top-level shape (single `error` key) keep working, while the
data is still preserved for those that look inside the error object.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(flows): always include the stopping step's result in early-stop errors

Drop the opt-in `error_include_result` gate. Since the step result is nested
inside the error object (`error.result`), the top-level result shape stays
`{ "error": .. }` — identical to a normal error — so consumers that detect or
parse failures by the top-level shape are unaffected. Gating it added schema
surface, plumbing, and a UI toggle for no real compatibility benefit.

Now, whenever a step early-stops with a raised error message, the flow fails
and the raised error embeds the stopping step's own result under
`error.result` (aggregated iteration results for loops/branchall). This
reverts the `StopAfterIf.error_include_result` field, its threading, the
OpenAPI/generated-client surface, and the editor toggle; the "Raise an error
message" tooltip now notes that the step result is included.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(flows): gate early-stop result inclusion behind opt-in flag

Re-introduce the per-step `error_include_result` flag (default off) instead
of always embedding the step result. Although nesting the result under
`error.result` keeps the result *shape* backward-compatible, it does not
address data exposure: a failed flow's result is propagated to synchronous
webhook callers, the flow's failure module, and the workspace/global error
handler (commonly a Slack/email/outbound-webhook notifier). Always including
the step output would surface previously-redacted intermediate data to all of
those sinks for every existing error-stop flow.

Gating keeps the existing behavior (bare `{ "error": .. }`) as the default and
only embeds `error.result` when the flow author explicitly opts in, matching
the original issue's intent.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(flows): omit error_include_result when false; refresh generated prompts

- Add `skip_serializing_if = "is_false"` to `StopAfterIf.error_include_result`
  so serialized flows are byte-identical when the flag is off. Fixes the
  `flowmodule_serde` round-trip test (cargo_test) and avoids churn on existing
  flows.
- Regenerate `system_prompts/auto-generated/` and `cli/src/guidance/skills.gen.ts`
  for the new OpenFlow `error_include_result` property. Fixes check-freshness.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(flows): cover error_include_result for the loop "stop after all iters" path

Add a regression test for the stop_after_all_iters_if branch, where `nresult`
already holds the aggregated iteration results — confirming `error.result`
carries each iteration's output (distinct from the per-step fallback path).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/tests/flow_engine_parity.rs

@@ -2916,6 +2916,7 @@ export function main() {
             expr: "flow_env.STOP === true".to_string(),
             skip_if_stopped: true,
             error_message: None,
+            error_include_result: false,
         });
         m
     };
@@ -2966,6 +2967,92 @@ export function main() {
     Ok(())
 }
 
+// stop_after_if with `error_message` + `error_include_result` should fail the
+// flow but preserve the stopping step's own result inside the raised error
+// object, i.e. `{ "error": { .., "result": <step result> } }`. With the flag off
+// (the default) the error object carries no `result`. Regression for the
+// early-stop branch in `update_flow_status_after_job_completion_internal`.
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base"))]
+async fn test_stop_after_if_error_include_result(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+
+    let make_flow = |include_result: bool| {
+        let mut m = flow_module(
+            "step",
+            FlowModuleValue::RawScript {
+                input_transforms: Default::default(),
+                language: ScriptLang::Deno,
+                content: r#"
+export function main() {
+    return { userErrors: ["email taken"], ok: false };
+}
+"#
+                .to_string(),
+                path: None,
+                lock: None,
+                tag: None,
+                concurrency_settings: Default::default(),
+                is_trigger: None,
+                assets: None,
+            },
+        );
+        m.stop_after_if = Some(windmill_common::flows::StopAfterIf {
+            expr: "true".to_string(),
+            skip_if_stopped: false,
+            error_message: Some("API returned userErrors".to_string()),
+            error_include_result: include_result,
+        });
+        FlowValue { modules: vec![m], same_worker: false, ..Default::default() }
+    };
+
+    // include_result = true: result preserves both the error and the step output
+    let job = RunJob::from(JobPayload::RawFlow {
+        value: make_flow(true),
+        path: None,
+        restarted_from: None,
+    })
+    .run_until_complete(&db, false, server.addr.port())
+    .await;
+    assert!(
+        !job.success,
+        "flow with raised early-stop error should fail"
+    );
+    let result = job.json_result().unwrap();
+    assert_eq!(
+        result["error"]["name"], "EarlyStopError",
+        "expected EarlyStopError; got {result:?}"
+    );
+    assert_eq!(result["error"]["message"], "API returned userErrors");
+    assert_eq!(
+        result["error"]["result"],
+        json!({ "userErrors": ["email taken"], "ok": false }),
+        "step result should be preserved under `error.result`; got {result:?}"
+    );
+
+    // include_result = false (default behavior): result is the bare error object
+    let job = RunJob::from(JobPayload::RawFlow {
+        value: make_flow(false),
+        path: None,
+        restarted_from: None,
+    })
+    .run_until_complete(&db, false, server.addr.port())
+    .await;
+    assert!(
+        !job.success,
+        "flow with raised early-stop error should fail"
+    );
+    let result = job.json_result().unwrap();
+    assert_eq!(result["error"]["name"], "EarlyStopError");
+    assert!(
+        result["error"].get("result").is_none(),
+        "without the flag the error must not embed the step result; got {result:?}"
+    );
+
+    Ok(())
+}
+
 // retry_if predicate sees flow_env. Regression for the two evaluate_retry
 // call sites in `update_flow_status_after_job_completion_internal` (lines
 // 1194 and 1576) which used to pass `None` for flow_env.
@@ -3093,6 +3180,7 @@ export function main(i: number) {
             expr: "flow_env.STOP === true".to_string(),
             skip_if_stopped: true,
             error_message: None,
+            error_include_result: false,
         });
         m
     };
@@ -3143,3 +3231,84 @@ export function main() {
 
     Ok(())
 }
+
+// stop_after_all_iters_if with `error_message` + `error_include_result` fails the
+// flow and embeds the loop's aggregated iteration results under `error.result`.
+// Covers the loop/branch-all path where `nresult` is already populated with the
+// aggregated results (distinct from the per-step fallback to `result`).
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base"))]
+async fn test_stop_after_all_iters_if_error_includes_result(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+
+    let inner = flow_module(
+        "iter_step",
+        FlowModuleValue::RawScript {
+            input_transforms: [js_input("i", "flow_input.iter.value")].into(),
+            language: ScriptLang::Deno,
+            content: r#"
+export function main(i: number) {
+    return { iter: i };
+}
+"#
+            .to_string(),
+            path: None,
+            lock: None,
+            tag: None,
+            concurrency_settings: Default::default(),
+            is_trigger: None,
+            assets: None,
+        },
+    );
+
+    let loop_module = {
+        let mut m = flow_module(
+            "loop",
+            FlowModuleValue::ForloopFlow {
+                iterator: InputTransform::Javascript { expr: "[1, 2, 3]".to_string() },
+                modules: vec![inner],
+                modules_node: None,
+                skip_failures: false,
+                parallel: false,
+                parallelism: None,
+                squash: None,
+            },
+        );
+        m.stop_after_all_iters_if = Some(windmill_common::flows::StopAfterIf {
+            expr: "true".to_string(),
+            skip_if_stopped: false,
+            error_message: Some("loop failed".to_string()),
+            error_include_result: true,
+        });
+        m
+    };
+
+    let flow = FlowValue { modules: vec![loop_module], same_worker: false, ..Default::default() };
+
+    let job = RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
+        .run_until_complete(&db, false, server.addr.port())
+        .await;
+
+    assert!(
+        !job.success,
+        "loop with a raised early-stop error should fail"
+    );
+    let result = job.json_result().unwrap();
+    assert_eq!(result["error"]["name"], "EarlyStopError", "got {result:?}");
+    assert_eq!(result["error"]["message"], "loop failed");
+    // error.result holds the aggregated iteration results (one per iteration)
+    let iters = result["error"]["result"].as_array().unwrap_or_else(|| {
+        panic!("error.result should be an array of iteration results; got {result:?}")
+    });
+    let iter_values: Vec<_> = iters.iter().map(|r| r["iter"].clone()).collect();
+    assert_eq!(
+        iter_values,
+        vec![json!(1), json!(2), json!(3)],
+        "error.result should contain each iteration's output; got {result:?}"
+    );
+
+    Ok(())
+}

backend/windmill-queue/src/jobs.rs

@@ -5292,6 +5292,7 @@ async fn push_inner<'c, 'd>(
                         expr: skip_handler.stop_condition,
                         skip_if_stopped: true,
                         error_message: Some(skip_handler.stop_message),
+                        error_include_result: false,
                     }),
                     ..Default::default()
                 });

backend/windmill-types/src/flows.rs

@@ -315,6 +315,11 @@ pub struct StopAfterIf {
     pub expr: String,
     pub skip_if_stopped: bool,
     pub error_message: Option<String>,
+    /// When stopping with an error (`error_message` set), embed the stopping
+    /// step's own result inside the raised error object (as `error.result`)
+    /// instead of discarding it. The top-level result stays `{ "error": .. }`.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub error_include_result: bool,
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]

backend/windmill-worker/src/worker_flow.rs

@@ -311,13 +311,14 @@ struct RecoveryObject {
     recover: Option<bool>,
 }
 
-fn get_stop_after_if_data(stop_after_if: Option<&StopAfterIf>) -> (bool, Option<String>) {
+/// Returns `(skip_if_stopped, error_message, include_step_result)`.
+fn get_stop_after_if_data(stop_after_if: Option<&StopAfterIf>) -> (bool, Option<String>, bool) {
     if let Some(stop_after_if) = stop_after_if {
         // skip_if_stopped and error_message are mutually exclusive:
         // skip_if_stopped=true means clean stop (mark remaining as skipped),
         // error_message means stop with error. skip_if_stopped takes precedence.
         if stop_after_if.skip_if_stopped {
-            return (true, None);
+            return (true, None, false);
         }
         let err_msg = stop_after_if.error_message.as_ref().and_then(|message| {
             if message.is_empty() {
@@ -326,9 +327,9 @@ fn get_stop_after_if_data(stop_after_if: Option<&StopAfterIf>) -> (bool, Option<
                 Some(message.clone())
             }
         });
-        return (false, err_msg);
+        return (false, err_msg, stop_after_if.error_include_result);
     }
-    return (false, None);
+    return (false, None, false);
 }
 
 async fn get_id_ctx_for_expr(
@@ -358,6 +359,7 @@ async fn evaluate_stop_after_all_iters_if(
     stop_early: &mut bool,
     skip_if_stop_early: &mut bool,
     stop_early_err_msg: &mut Option<String>,
+    stop_early_include_result: &mut bool,
     nresult: &mut Option<Arc<Box<RawValue>>>,
     args: HashMap<String, Box<RawValue>>,
     flow_env: Option<&HashMap<String, Box<RawValue>>>,
@@ -394,8 +396,11 @@ async fn evaluate_stop_after_all_iters_if(
 
     if stop_early_after_all_iters {
         *stop_early = true;
-        (*skip_if_stop_early, *stop_early_err_msg) =
-            get_stop_after_if_data(Some(stop_after_all_iters_if));
+        (
+            *skip_if_stop_early,
+            *stop_early_err_msg,
+            *stop_early_include_result,
+        ) = get_stop_after_if_data(Some(stop_after_all_iters_if));
     }
     Ok(())
 }
@@ -655,19 +660,24 @@ pub async fn update_flow_status_after_job_completion_internal(
             false
         };
 
-        let (mut stop_early, mut stop_early_err_msg, mut skip_if_stop_early, continue_on_error) =
-            if stop_early_override.is_some()
-                && !is_flow_stop_early_override
-                && !parallel_loop
-                && !parallel_branchall
-            {
-                // we ignore stop_early_override (stop_early in children) if module is parallel or is a flow step
-                let se = stop_early_override.as_ref().unwrap();
-                (true, None, *se, false)
-            } else if is_failure_step || module_step.is_preprocessor_step() {
-                (false, None, false, false)
-            } else if let Some(current_module) = current_module {
-                let stop_early = success
+        let (
+            mut stop_early,
+            mut stop_early_err_msg,
+            mut skip_if_stop_early,
+            mut stop_early_include_result,
+            continue_on_error,
+        ) = if stop_early_override.is_some()
+            && !is_flow_stop_early_override
+            && !parallel_loop
+            && !parallel_branchall
+        {
+            // we ignore stop_early_override (stop_early in children) if module is parallel or is a flow step
+            let se = stop_early_override.as_ref().unwrap();
+            (true, None, *se, false, false)
+        } else if is_failure_step || module_step.is_preprocessor_step() {
+            (false, None, false, false, false)
+        } else if let Some(current_module) = current_module {
+            let stop_early = success
                     && !is_branch_all // we don't support stop_early per branch
                     && !parallel_loop // we don't support anymore stop_early per iteration when parallel for loop (removed from frontend)
                     && !is_identity_job // don't evaluate stop_after_if for skipped (identity) steps
@@ -717,22 +727,23 @@ pub async fn update_flow_status_after_job_completion_internal(
                     } else {
                         false
                     };
-                let (skip_if_stopped, stop_early_err_msg) = if stop_early {
-                    get_stop_after_if_data(current_module.stop_after_if.as_ref())
-                } else {
-                    (false, None)
-                };
-
-                (
-                    stop_early,
-                    stop_early_err_msg,
-                    skip_if_stopped,
-                    current_module.continue_on_error.unwrap_or(false),
-                )
+            let (skip_if_stopped, stop_early_err_msg, include_result) = if stop_early {
+                get_stop_after_if_data(current_module.stop_after_if.as_ref())
             } else {
-                (false, None, false, false)
+                (false, None, false)
             };
 
+            (
+                stop_early,
+                stop_early_err_msg,
+                skip_if_stopped,
+                include_result,
+                current_module.continue_on_error.unwrap_or(false),
+            )
+        } else {
+            (false, None, false, false, false)
+        };
+
         let skip_seq_branch_failure = match module_status {
             FlowStatusModule::InProgress {
                 branchall: Some(BranchAllStatus { branch, .. }),
@@ -974,6 +985,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             &mut stop_early,
                             &mut skip_if_stop_early,
                             &mut stop_early_err_msg,
+                            &mut stop_early_include_result,
                             &mut nresult,
                             args,
                             resolved_flow_env.as_deref(),
@@ -1173,6 +1185,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                     stop_early = false;
                     stop_early_err_msg = None;
                     skip_if_stop_early = false;
+                    stop_early_include_result = false;
                 }
 
                 if is_loop || (is_branch_all && !stop_early) {
@@ -1194,6 +1207,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             &mut stop_early,
                             &mut skip_if_stop_early,
                             &mut stop_early_err_msg,
+                            &mut stop_early_include_result,
                             &mut nresult,
                             args,
                             resolved_flow_env.as_deref(),
@@ -1310,12 +1324,22 @@ pub async fn update_flow_status_after_job_completion_internal(
         };
 
         if stop_early && stop_early_err_msg.is_some() {
-            nresult = Some(Arc::new(to_raw_value(&serde_json::json! ({
-                "error": {
-                    "name": "EarlyStopError",
-                    "message": stop_early_err_msg.as_ref().unwrap(),
-                }
-            }))));
+            let mut error = serde_json::json!({
+                "name": "EarlyStopError",
+                "message": stop_early_err_msg.as_ref().unwrap(),
+            });
+            if stop_early_include_result {
+                // Embed the stopping step's own result inside the error object instead
+                // of discarding it, keeping the top-level result shape `{ "error": .. }`
+                // unchanged. `nresult` is already set for loops/branchall (aggregated
+                // iteration results), otherwise fall back to the step result.
+                let step_result = nresult.clone().unwrap_or_else(|| result.clone());
+                error["result"] =
+                    serde_json::to_value(&step_result).unwrap_or(serde_json::Value::Null);
+            }
+            nresult = Some(Arc::new(to_raw_value(
+                &serde_json::json!({ "error": error }),
+            )));
         }
 
         let step_counter = if inc_step_counter {

frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte

@@ -99,7 +99,8 @@
 						flowModule.stop_after_if = {
 							expr: 'result == undefined',
 							skip_if_stopped: false,
-							error_message: undefined
+							error_message: undefined,
+							error_include_result: false
 						}
 					}
 				}}
@@ -137,6 +138,7 @@
 								on:change={(event) => {
 									if (flowModule.stop_after_if && event.detail) {
 										flowModule.stop_after_if.error_message = undefined
+										flowModule.stop_after_if.error_include_result = false
 										raise_error_message_stop_after_if = false
 									}
 								}}
@@ -154,6 +156,7 @@
 											flowModule.stop_after_if.skip_if_stopped = false
 										} else {
 											flowModule.stop_after_if.error_message = undefined
+											flowModule.stop_after_if.error_include_result = false
 										}
 									}
 								}}
@@ -171,6 +174,15 @@
 							bind:value={flowModule.stop_after_if.error_message}
 							placeholder="Enter custom error message (optional)"
 						/>
+						<Toggle
+							size="xs"
+							bind:checked={flowModule.stop_after_if.error_include_result}
+							options={{
+								right: "Include the stopping step's result in the error",
+								rightTooltip:
+									"When enabled, this step's output is embedded inside the raised error object (as error.result) instead of being discarded. The flow result stays { error }."
+							}}
+						/>
 					{/if}
 					<span class="mt-2 text-xs font-bold">Stop condition expression</span>
 					<div class="border rounded-md w-full overflow-auto">
@@ -253,7 +265,8 @@
 						flowModule.stop_after_all_iters_if = {
 							expr: 'result == undefined',
 							skip_if_stopped: false,
-							error_message: undefined
+							error_message: undefined,
+							error_include_result: false
 						}
 					}
 				}}
@@ -283,6 +296,7 @@
 								on:change={(event) => {
 									if (flowModule.stop_after_all_iters_if && event.detail) {
 										flowModule.stop_after_all_iters_if.error_message = undefined
+										flowModule.stop_after_all_iters_if.error_include_result = false
 										raise_error_message_stop_after_all_if = false
 									}
 								}}
@@ -300,6 +314,7 @@
 											flowModule.stop_after_all_iters_if.skip_if_stopped = false
 										} else {
 											flowModule.stop_after_all_iters_if.error_message = undefined
+											flowModule.stop_after_all_iters_if.error_include_result = false
 										}
 									}
 								}}
@@ -317,6 +332,15 @@
 							bind:value={flowModule.stop_after_all_iters_if.error_message}
 							placeholder="Enter custom error message (optional)"
 						/>
+						<Toggle
+							size="xs"
+							bind:checked={flowModule.stop_after_all_iters_if.error_include_result}
+							options={{
+								right: "Include the stopping step's result in the error",
+								rightTooltip:
+									"When enabled, this step's output is embedded inside the raised error object (as error.result) instead of being discarded. The flow result stays { error }."
+							}}
+						/>
 					{/if}
 					<span class="mt-2 text-xs font-bold">Stop condition expression</span>
 					<div class="border rounded-md w-full overflow-auto">

openflow.openapi.yaml

@@ -271,6 +271,9 @@ components:
           type: string
           nullable: true
           description: Custom error message when stopping with an error. Mutually exclusive with skip_if_stopped. If set to a non-empty string, the flow stops with this error. If empty string, a default error message is used. If null or omitted, no error is raised.
+        error_include_result:
+          type: boolean
+          description: "When stopping with an error (error_message set), embed the stopping step's own result inside the raised error object (as error.result) instead of discarding it. The top-level result stays { error }. Defaults to false."
       required:
         - expr