feat: support $flow_expr[...] dynamic tags on flow steps (#11170)

* feat: resolve $flow_expr[...] dynamic tags on flow steps Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * 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> * refactor: resolve $flow_expr tags by path lookup instead of expression evaluation Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * docs: say a $flow_expr tag fails to resolve, not to evaluate Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(frontend): word the $flow_expr help like the other dynamic tag lines Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix: reject a malformed $flow_expr placeholder instead of queueing its literal tag Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * refactor: render $flow_expr tag values through the $args path lookup Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * test: pin interpolate_args through the shared tag path rendering Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-09-21 00:02:30 +00:00 · 2026-09-18 14:37:50 +02:00
parent 9320312eac
commit 48f00259c5
4 changed files with 259 additions and 50 deletions
@@ -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(())
+}
@@ -4650,6 +4650,30 @@ 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[")
+}
+
+/// Renders the value at the dotted `path` below `root` as a dynamic tag component, shared by
+/// `$args[...]` and `$flow_expr[...]`: its JSON text with surrounding quotes trimmed, and empty
+/// once a segment is missing. Only object keys are followed, never array indexes.
+pub fn render_tag_path(root: Option<&RawValue>, path: &str) -> String {
+    let mut value = root.map(|x| x.get()).unwrap_or_default().to_string();
+    for part in path.split('.').filter(|p| !p.is_empty()) {
+        match serde_json::from_str::<serde_json::Value>(&value) {
+            Ok(obj) => value = obj.get(part).map(|v| v.to_string()).unwrap_or_default(),
+            Err(_) => {
+                value = String::new();
+                break;
+            }
+        }
+    }
+    value.trim_matches('"').to_string()
+}
+
 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();
@@ -4657,40 +4681,12 @@ pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> Strin
        let mut interpolated = workspaced.clone();
        for cap in RE_ARG_TAG.captures_iter(&workspaced) {
            let arg_name = cap.get(1).unwrap().as_str();
-            let arg_value = if arg_name.contains('.') {
-                let parts: Vec<&str> = arg_name.split('.').collect();
-                let root = parts[0];
-                let mut value = args
-                    .args
-                    .get(root)
-                    .or(args.extra.as_ref().and_then(|x| x.get(root)))
-                    .map(|x| x.get())
-                    .unwrap_or_default()
-                    .to_string();
-
-                for part in parts.iter().skip(1) {
-                    if let Ok(obj) = serde_json::from_str::<serde_json::Value>(&value) {
-                        value = obj
-                            .get(part)
-                            .and_then(|v| Some(v.to_string()))
-                            .unwrap_or_default()
-                            .as_str()
-                            .to_string();
-                    } else {
-                        value = "".to_string(); // Invalid JSON or missing field
-                        break;
-                    }
-                }
-                value.trim_matches('"').to_string()
-            } else {
-                args.args
-                    .get(arg_name)
-                    .or(args.extra.as_ref().and_then(|x| x.get(arg_name)))
-                    .map(|x| x.get())
-                    .unwrap_or_default()
-                    .trim_matches('"')
-                    .to_string()
-            };
+            let (root, rest) = arg_name.split_once('.').unwrap_or((arg_name, ""));
+            let root_value = args
+                .args
+                .get(root)
+                .or(args.extra.as_ref().and_then(|x| x.get(root)));
+            let arg_value = render_tag_path(root_value.map(|x| &**x), rest);
            interpolated =
                interpolated.replace(format!("$args[{}]", arg_name).as_str(), &arg_value);
        }
@@ -5269,6 +5265,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 +6535,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;
        }

@@ -7922,3 +7923,43 @@ mod result_metadata_tests {
        assert_eq!(meta.wm_failure.as_deref(), Some("boom"));
    }
 }
+
+#[cfg(test)]
+mod render_tag_path_tests {
+    use super::{interpolate_args, render_tag_path, PushArgs};
+    use serde_json::value::RawValue;
+    use std::collections::HashMap;
+
+    fn render(root: &str, path: &str) -> String {
+        render_tag_path(
+            Some(&RawValue::from_string(root.to_string()).unwrap()),
+            path,
+        )
+    }
+
+    // Existing `$args[...]` tags route on exactly these renderings.
+    #[test]
+    fn renders_like_args_tags() {
+        assert_eq!(render(r#""eu""#, ""), "eu");
+        assert_eq!(render(r#"{"a": {"b": "eu"}}"#, "a.b"), "eu");
+        assert_eq!(render(r#"{"n": 4}"#, "n"), "4");
+        assert_eq!(render("null", ""), "null");
+        assert_eq!(render(r#"{"a": 1}"#, "b.c"), "");
+        assert_eq!(render(r#"{"a": ["eu"]}"#, "a.0"), "");
+        assert_eq!(render_tag_path(None, "a"), "");
+
+        let args = HashMap::from([("cfg".to_string(), raw(r#"{"lang": "eu"}"#))]);
+        let push_args = PushArgs {
+            args: &args,
+            extra: Some(HashMap::from([("e".to_string(), raw(r#""x""#))])),
+        };
+        assert_eq!(
+            interpolate_args("w-$args[cfg.lang]-$args[e]".to_string(), &push_args, "ws"),
+            "w-eu-x"
+        );
+    }
+
+    fn raw(json: &str) -> Box<RawValue> {
+        RawValue::from_string(json.to_string()).unwrap()
+    }
+}
@@ -69,10 +69,10 @@ use windmill_common::{
 use windmill_queue::schedule::get_schedule_opt;
 use windmill_queue::{
    add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
-    insert_concurrency_key_capped, interpolate_args,
+    insert_concurrency_key_capped, interpolate_args, render_tag_path,
    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;
@@ -3115,6 +3115,66 @@ fn resolve_flow_step_tag(
    }
 }

+/// Resolves each `$flow_expr[root.key.path]` of a step tag by reading `key.path` from `results`
+/// (where `key` is a step id), `flow_input` or `flow_env`, rendered as `$args[key.path]` would be.
+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 (root, key_path) = path.split_once('.').unwrap_or((path, ""));
+        let (key, rest) = key_path.split_once('.').unwrap_or((key_path, ""));
+        if key.is_empty() || !matches!(root, "results" | "flow_input" | "flow_env") {
+            return Err(Error::ExecutionErr(format!(
+                "Could not resolve the step tag `{tag}`: `{path}` must start with \
+                 `results.<step_id>`, `flow_input.<key>` or `flow_env.<key>`"
+            )));
+        }
+        let value = match root {
+            "flow_input" => render_tag_path(flow_input.get(key).map(|x| &**x), rest),
+            "flow_env" => render_tag_path(flow_env.and_then(|e| e.get(key)).map(|x| &**x), rest),
+            _ => match windmill_queue::get_result_by_id(
+                db.clone(),
+                flow_job.workspace_id.clone(),
+                flow_job.id,
+                key.to_string(),
+                None,
+            )
+            .await
+            {
+                Ok(result) => render_tag_path(Some(&*result), rest),
+                Err(Error::NotFound(_)) => String::new(),
+                Err(e) => {
+                    return Err(Error::ExecutionErr(format!(
+                        "Could not resolve the step tag `{tag}`: {e}"
+                    )))
+                }
+            },
+        };
+        rendered.insert(path, value);
+    }
+    Ok(RE_FLOW_EXPR_TAG
+        .replace_all(tag, |cap: &regex::Captures| rendered[&cap[1]].clone())
+        .into_owned())
+}
+
 #[cfg(test)]
 mod tag_resolution_tests {
    use super::resolve_flow_step_tag;
@@ -4193,6 +4253,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 {
@@ -4217,8 +4281,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,
@@ -4397,17 +4463,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 {
@@ -4449,6 +4519,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);

@@ -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>