fix: reject a malformed $flow_expr placeholder instead of queueing its literal tag

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

backend/tests/flow_engine_parity.rs

@@ -3312,3 +3312,60 @@ export function main(i: number) {
 
     Ok(())
 }
+
+// A `$flow_expr[...]` step tag is resolved from the flow's state before the step is pushed, and
+// one that cannot be resolved fails the step instead of queueing it on a tag no worker serves.
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base"))]
+async fn test_flow_expr_step_tag(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+
+    let step = |id: &str, tag: Option<&str>| {
+        flow_module(
+            id,
+            FlowModuleValue::RawScript {
+                input_transforms: Default::default(),
+                language: ScriptLang::Deno,
+                content: "export function main() { return { lang: 'bun' } }".to_string(),
+                path: None,
+                lock: None,
+                tag: tag.map(str::to_string),
+                concurrency_settings: Default::default(),
+                is_trigger: None,
+                assets: None,
+            },
+        )
+    };
+    let flow = FlowValue {
+        modules: vec![
+            step("a", None),
+            step("b", Some("$flow_expr[results.a.lang]")),
+            step("c", Some("nobody-serves-$flow_expr[a.lang]")),
+        ],
+        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;
+
+    let b_tag = sqlx::query_scalar::<_, String>(
+        "SELECT tag FROM v2_job WHERE parent_job = $1 AND flow_step_id = 'b'",
+    )
+    .bind(job.id)
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(b_tag, "bun");
+
+    assert!(!job.success);
+    let result = job.json_result().unwrap();
+    let message = result["error"]["message"].as_str().unwrap_or_default();
+    assert!(
+        message.contains("Could not resolve the step tag `nobody-serves-$flow_expr[a.lang]`"),
+        "got {result:?}"
+    );
+
+    Ok(())
+}

backend/windmill-queue/src/jobs.rs

@@ -4650,6 +4650,13 @@ 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]`), which only the flow
+/// runtime can resolve, right before pushing the step. A malformed placeholder counts too, so it
+/// is rejected or dropped instead of queueing the job on its literal text.
+pub fn tag_reads_flow_expr(tag: &str) -> bool {
+    tag.contains("$flow_expr[")
+}
+
 pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
     // Save this value to avoid parsing twice
     let workspaced = x.as_str().replace("$workspace", workspace_id).to_string();
@@ -5269,6 +5276,8 @@ pub fn empty_result() -> Box<RawValue> {
 
 lazy_static::lazy_static! {
     pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[((?:\w+\.)*\w+)\]"#).unwrap();
+    pub static ref RE_FLOW_EXPR_TAG: Regex =
+        Regex::new(r#"\$flow_expr\[((?:\w+\.)*\w+)\]"#).unwrap();
 }
 
 #[cfg(feature = "cloud")]
@@ -6537,7 +6546,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;
         }
 

backend/windmill-worker/src/worker_flow.rs

@@ -71,8 +71,8 @@ use windmill_queue::{
     add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
     insert_concurrency_key_capped, interpolate_args,
     report_error_to_workspace_handler_or_critical_side_channel, tag_reads_args,
-    try_schedule_next_job, CanceledBy, FlowRunners, MiniCompletedJob, MiniPulledJob, PushArgs,
-    PushIsolationLevel, SameWorkerPayload, WrappedError,
+    tag_reads_flow_expr, try_schedule_next_job, CanceledBy, FlowRunners, MiniCompletedJob,
+    MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError, RE_FLOW_EXPR_TAG,
 };
 
 use windmill_audit::audit_oss::audit_log;
@@ -3114,6 +3114,88 @@ fn resolve_flow_step_tag(
     }
 }
 
+/// Resolves each `$flow_expr[root.path]` of a step tag by reading `path` from `results` (keyed by
+/// step id), `flow_input` or `flow_env`. As for `$args[...]`, a path that reaches nothing renders
+/// empty, a string renders bare and any other value as its JSON text.
+async fn interpolate_flow_expr_tag(
+    tag: &str,
+    db: &DB,
+    flow_job: &MiniPulledJob,
+    flow_input: &HashMap<String, Box<RawValue>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
+) -> error::Result<String> {
+    if RE_FLOW_EXPR_TAG
+        .replace_all(tag, "")
+        .contains("$flow_expr[")
+    {
+        return Err(Error::ExecutionErr(format!(
+            "Could not resolve the step tag `{tag}`: each `$flow_expr[...]` must hold a dotted \
+             path such as `results.a.b.c`"
+        )));
+    }
+    let mut rendered: HashMap<&str, String> = HashMap::new();
+    for cap in RE_FLOW_EXPR_TAG.captures_iter(tag) {
+        let path = cap.get(1).unwrap().as_str();
+        if rendered.contains_key(path) {
+            continue;
+        }
+        let mut segments = path.split('.');
+        let root = segments.next().unwrap_or_default();
+        let segments = segments.collect::<Vec<_>>();
+        let value = match (root, segments.split_first()) {
+            ("flow_input", _) => read_flow_expr_path(Some(flow_input), &segments),
+            ("flow_env", _) => read_flow_expr_path(flow_env, &segments),
+            ("results", Some((step_id, rest))) => {
+                match windmill_queue::get_result_by_id(
+                    db.clone(),
+                    flow_job.workspace_id.clone(),
+                    flow_job.id,
+                    step_id.to_string(),
+                    (!rest.is_empty()).then(|| rest.join(".")),
+                )
+                .await
+                {
+                    Ok(result) => serde_json::from_str(result.get()).unwrap_or_default(),
+                    Err(Error::NotFound(_)) => Value::Null,
+                    Err(e) => {
+                        return Err(Error::ExecutionErr(format!(
+                            "Could not resolve the step tag `{tag}`: {e}"
+                        )))
+                    }
+                }
+            }
+            _ => {
+                return Err(Error::ExecutionErr(format!(
+                    "Could not resolve the step tag `{tag}`: `{path}` must start with \
+                     `results.<step_id>`, `flow_input` or `flow_env`"
+                )))
+            }
+        };
+        let value = match value {
+            Value::String(s) => s,
+            Value::Null => String::new(),
+            v => v.to_string(),
+        };
+        rendered.insert(path, value);
+    }
+    Ok(RE_FLOW_EXPR_TAG
+        .replace_all(tag, |cap: &regex::Captures| rendered[&cap[1]].clone())
+        .into_owned())
+}
+
+fn read_flow_expr_path(map: Option<&HashMap<String, Box<RawValue>>>, segments: &[&str]) -> Value {
+    let Some((key, rest)) = segments.split_first() else {
+        return Value::Null;
+    };
+    map.and_then(|m| m.get(*key))
+        .and_then(|raw| serde_json::from_str::<Value>(raw.get()).ok())
+        .and_then(|v| {
+            v.pointer(&rest.iter().map(|s| format!("/{s}")).collect::<String>())
+                .cloned()
+        })
+        .unwrap_or_default()
+}
+
 #[cfg(test)]
 mod tag_resolution_tests {
     use super::resolve_flow_step_tag;
@@ -4192,6 +4274,10 @@ async fn push_next_flow_job(
             None
         };
 
+        // The `flow_input` the step's input transforms read, which a `$flow_expr[flow_input...]`
+        // tag must read too: the body of a simple for-loop also sees `iter` there.
+        let mut step_flow_input = arc_flow_job_args.clone();
+
         let marc;
         let me;
         let args = match &next_status {
@@ -4216,8 +4302,10 @@ async fn push_next_flow_job(
                 if let Some(input_transforms) = simple_input_transforms {
                     //previous id is none because we do not want to use previous id if we are in a for loop
                     let ctx = get_transform_context(&flow_job, "", &status);
+                    let args = Marc::new(args);
+                    step_flow_input = args.clone();
                     let ti = transform_input(
-                        Marc::new(args),
+                        args,
                         flow_env,
                         arc_last_job_result.clone(),
                         input_transforms,
@@ -4396,17 +4484,21 @@ async fn push_next_flow_job(
             payload_tag.tag.as_deref(),
         );
 
-        // `push_args` is empty once the input transforms failed, so a tag reading `$args[...]`
-        // interpolates to a queue nobody serves and the step sits there instead of reporting
-        // the error. Send it to the flow's tag, which a worker is provably serving right now.
+        // A step whose inputs failed to evaluate, or whose `$flow_expr[...]` tag failed to resolve,
+        // is pushed only to report the error, and a computed tag can then name a queue nobody
+        // serves (`push_args` is empty, so `$args[...]` reads nothing), leaving the step stuck
+        // instead. Send it to the flow's tag, which a worker is provably serving right now.
         //
         // A step handed over by id, or one whose tag `push` replaces, never reaches a worker
         // through its tag, so rewriting theirs would be noise.
         let step_is_pulled_by_tag = !continue_on_same_worker
             && !continue_with_runners
             && !payload_tag.payload.is_dedicated_worker();
-        let reroute_to_flow_tag =
-            err.is_some() && step_is_pulled_by_tag && tag.as_deref().is_some_and(tag_reads_args);
+        let reroute_to_flow_tag = err.is_some()
+            && step_is_pulled_by_tag
+            && tag
+                .as_deref()
+                .is_some_and(|t| tag_reads_args(t) || tag_reads_flow_expr(t));
         let tag = if reroute_to_flow_tag {
             Some(flow_job.tag.clone())
         } else {
@@ -4448,6 +4540,30 @@ async fn push_next_flow_job(
             .await?;
         }
 
+        // Resolved only after the check: CUSTOM_TAGS allows the template, so its value may name
+        // any queue, as the value of an `$args[...]` tag does.
+        let mut tag_err = None;
+        let tag = match tag {
+            Some(t) if err.is_none() && tag_reads_flow_expr(&t) => {
+                match interpolate_flow_expr_tag(&t, db, &flow_job, &step_flow_input, flow_env)
+                    .warn_after_seconds(3)
+                    .await
+                {
+                    Ok(resolved) => Some(resolved),
+                    Err(e) => {
+                        tag_err = Some(e);
+                        Some(if step_is_pulled_by_tag {
+                            flow_job.tag.clone()
+                        } else {
+                            t
+                        })
+                    }
+                }
+            }
+            t => t,
+        };
+        let err = err.or(tag_err.as_ref());
+
         let evaluated_timeout = if let Some(timeout_transform) = &module.timeout {
             let ctx = get_transform_context(&flow_job, &previous_id, &status);
 

frontend/src/lib/components/AssignableTagsInner.svelte

@@ -39,7 +39,8 @@
 
 	// Mirrors CUSTOM_TAG_REGEX in backend/windmill-common/src/worker.rs — keep both in sync.
 	const customTagRegex = /^([\w-]+)\(((?:[\w-]+\*?\+)*[\w-]+\*?|(?:\^[\w-]+\*?)+)\)$/
-	const dynamicTagRegex = /\$args\[((?:\w+\.)*\w+)\]/
+	// Mirrors RE_ARG_TAG and RE_FLOW_EXPR_TAG in backend/windmill-queue/src/jobs.rs.
+	const dynamicTagRegex = /\$(args|flow_expr)\[((?:\w+\.)*\w+)\]/
 
 	function formatWorkspace(w: { id: string; includeForks: boolean }) {
 		return w.includeForks ? `${w.id} (and its forks)` : w.id
@@ -49,7 +50,7 @@
 		let r = newTag.trim()
 		if (r == '') return undefined
 		let matched = r.match(dynamicTagRegex)
-		return matched?.[1]
+		return matched ? { kind: matched[1], path: matched[2] } : undefined
 	})
 
 	let extractedCustomTag = $derived.by(() => {
@@ -183,7 +184,7 @@
 				</div>
 			</div>
 		{:else if newTag.trim()}
-			{#if newTag.includes('(') || newTag.includes(')') || newTag.includes('+') || newTag.includes('^') || newTag.includes('*') || ((newTag.includes('.') || newTag.includes('$args[')) && !dynamicTag)}
+			{#if newTag.includes('(') || newTag.includes(')') || newTag.includes('+') || newTag.includes('^') || newTag.includes('*') || ((newTag.includes('.') || newTag.includes('$args[') || newTag.includes('$flow_expr[')) && !dynamicTag)}
 				<div class="text-2xs text-primary p-2 bg-surface-secondary rounded border">
 					<div class="font-medium mb-1 text-red-500">Invalid tag</div>
 					<div>
@@ -194,7 +195,7 @@
 			{:else}
 				<div class="text-2xs text-primary p-2 bg-surface-secondary rounded border">
 					<div class="font-medium mb-1">
-						{#if newTag.includes('$workspace') || newTag.includes('$args')}
+						{#if newTag.includes('$workspace') || dynamicTag}
 							Dynamic tag
 						{:else}
 							Simple tag
@@ -207,8 +208,13 @@
 					{#if newTag.includes('$workspace') && !dynamicTag}
 						<div>Interpolated tag based on workspace id the job was created in </div>
 					{/if}
-					{#if dynamicTag}
-						<div>Interpolated tag based on args input of <b>{dynamicTag}</b></div>
+					{#if dynamicTag?.kind == 'flow_expr'}
+						<div>
+							Interpolated tag based on the flow value at <b>{dynamicTag.path}</b>, resolved when
+							the flow step starts
+						</div>
+					{:else if dynamicTag}
+						<div>Interpolated tag based on args input of <b>{dynamicTag.path}</b></div>
 					{/if}
 				</div>
 			{/if}
@@ -252,6 +258,17 @@
 			>
 			based on args input, use <pre class="inline text-emphasis">$args[a.b.c]</pre> where
 			<pre class="inline">a.b.c</pre> is the path to the value in the args object.
+			<br />{#if variant !== 'drawer'}<br />{/if}
+			For
+			<a
+				href="https://www.windmill.dev/docs/core_concepts/worker_groups#dynamic-tag"
+				target="_blank">dynamic tags <ExternalLink size={12} class="inline-block" /></a
+			>
+			based on flow step results, flow input or flow env, use
+			<pre class="inline text-emphasis">$flow_expr[results.a.b.c]</pre> where
+			<pre class="inline">a</pre> is the step id, or
+			<pre class="inline text-emphasis">$flow_expr[flow_input.a.b.c]</pre> and
+			<pre class="inline text-emphasis">$flow_expr[flow_env.a.b.c]</pre>.
 		</span>
 	{/if}
 </div>