From 1731e518be9d67ece0c211a22221d05bac3e0f6b Mon Sep 17 00:00:00 2001
From: hugocasa <hugo@casademont.ch>
Date: Wed, 16 Sep 2026 13:54:29 +0200
Subject: [PATCH] fix: run an unresolvable $flow_expr tag on the default tag
 and give it the step input scope

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 backend/windmill-queue/src/jobs.rs         |  9 ++++---
 backend/windmill-worker/src/worker_flow.rs | 29 ++++++++++++++++------
 2 files changed, 27 insertions(+), 11 deletions(-)

diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index 6a99082596..cb741f35ac 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -4650,8 +4650,8 @@ pub fn tag_reads_args(tag: &str) -> bool {
     RE_ARG_TAG.is_match(tag)
 }
 
-/// Whether the tag reads the flow's state (`$flow_expr[results.a.foo]`). Only the flow runtime
-/// can resolve it, right before pushing the step; `push` leaves it verbatim.
+/// Whether the tag reads the flow's state (`$flow_expr[results.a.foo]`), which only the flow
+/// runtime can resolve, right before pushing the step.
 pub fn tag_reads_flow_expr(tag: &str) -> bool {
     RE_FLOW_EXPR_TAG.is_match(tag)
 }
@@ -6545,7 +6545,10 @@ async fn push_inner<'c, 'd>(
         );
         windmill_common::worker::dedicated_worker_tag(workspace_id, &full_path)
     } else {
-        if tag == Some("".to_string()) {
+        // The flow runtime resolves a step's `$flow_expr[...]` before pushing it, so one still here
+        // was pushed with no flow state to read (a step test, a dependency job) and would name a
+        // queue no worker serves: the job runs on its default tag instead.
+        if tag == Some("".to_string()) || tag.as_deref().is_some_and(tag_reads_flow_expr) {
             tag = None;
         }
 
diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 3b0d600607..e58f0357b3 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -3115,11 +3115,12 @@ fn resolve_flow_step_tag(
 }
 
 /// Resolves each `$flow_expr[path]` of a step tag by evaluating `path` as a flow expression
-/// (`results.a.foo`, `flow_input.region`, `flow_env.pool`). A string renders bare, `null` (also
-/// what a missing field evaluates to) renders empty, any other value as its JSON text.
+/// (`results.a.foo`, `flow_input.region`, `flow_env.pool`, `resume.region`). A string renders
+/// bare, `null` (also what a missing field evaluates to) renders empty, any other value as its
+/// JSON text.
 async fn interpolate_flow_expr_tag(
     tag: &str,
-    last_result: Arc<Box<RawValue>>,
+    env: HashMap<String, Arc<Box<RawValue>>>,
     flow_args: Marc<HashMap<String, Box<RawValue>>>,
     flow_env: Option<&HashMap<String, Box<RawValue>>>,
     client: &AuthedClient,
@@ -3131,16 +3132,22 @@ async fn interpolate_flow_expr_tag(
         if rendered.contains_key(expr) {
             continue;
         }
-        let value = evaluate_input_transform::<Value>(
-            &InputTransform::new_javascript_expr(expr),
-            last_result.clone(),
+        let value = eval_timeout(
+            expr.to_string(),
+            env.clone(),
             Some(flow_args.clone()),
             flow_env,
             Some(client),
             Some(by_id),
+            None,
         )
         .await
-        .map_err(|e| Error::ExecutionErr(format!("Could not resolve the step tag `{tag}`: {e}")))?;
+        .map_err(|e| {
+            Error::ExecutionErr(format!(
+                "Could not resolve the step tag `{tag}`: error during evaluation of `{expr}`:\n{e:#}"
+            ))
+        })?;
+        let value = serde_json::from_str::<Value>(value.get()).map_err(to_anyhow)?;
         rendered.insert(expr, render_flow_expr_tag_value(value));
     }
     Ok(RE_FLOW_EXPR_TAG
@@ -4508,9 +4515,15 @@ async fn push_next_flow_job(
         let tag = match tag {
             Some(t) if err.is_none() && tag_reads_flow_expr(&t) => {
                 let ctx = get_transform_context(&flow_job, expr_previous_id, &status);
+                let env = HashMap::from([
+                    ("previous_result".to_string(), arc_last_job_result.clone()),
+                    ("resume".to_string(), resume.clone()),
+                    ("resumes".to_string(), resumes.clone()),
+                    ("approvers".to_string(), approvers.clone()),
+                ]);
                 match interpolate_flow_expr_tag(
                     &t,
-                    arc_last_job_result.clone(),
+                    env,
                     expr_flow_args.clone(),
                     flow_env,
                     client,