From 48f00259c5e7361d3553dbcb809461e1cde96f8f Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 18 Sep 2026 14:37:50 +0200 Subject: [PATCH] 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 * fix: run an unresolvable $flow_expr tag on the default tag and give it the step input scope Co-Authored-By: Claude Opus 5 * refactor: resolve $flow_expr tags by path lookup instead of expression evaluation Co-Authored-By: Claude Opus 5 * docs: say a $flow_expr tag fails to resolve, not to evaluate Co-Authored-By: Claude Opus 5 * fix(frontend): word the $flow_expr help like the other dynamic tag lines Co-Authored-By: Claude Opus 5 * fix: reject a malformed $flow_expr placeholder instead of queueing its literal tag Co-Authored-By: Claude Opus 5 * refactor: render $flow_expr tag values through the $args path lookup Co-Authored-By: Claude Opus 5 * test: pin interpolate_args through the shared tag path rendering Co-Authored-By: Claude Opus 5 --------- Co-authored-by: Claude Opus 5 --- backend/tests/flow_engine_parity.rs | 57 +++++++++ backend/windmill-queue/src/jobs.rs | 111 +++++++++++------ backend/windmill-worker/src/worker_flow.rs | 112 ++++++++++++++++-- .../lib/components/AssignableTagsInner.svelte | 29 ++++- 4 files changed, 259 insertions(+), 50 deletions(-) diff --git a/backend/tests/flow_engine_parity.rs b/backend/tests/flow_engine_parity.rs index 78e45e404b..b6a943a23b 100644 --- a/backend/tests/flow_engine_parity.rs +++ b/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) -> 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(()) +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ab91d54287..429701f7e0 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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::(&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::(&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 { 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::from_string(json.to_string()).unwrap() + } +} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 6c0dcaf279..dbdba3e6ed 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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>, + flow_env: Option<&HashMap>>, +) -> error::Result { + 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.`, `flow_input.` or `flow_env.`" + ))); + } + 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: ®ex::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); diff --git a/frontend/src/lib/components/AssignableTagsInner.svelte b/frontend/src/lib/components/AssignableTagsInner.svelte index e1ad0baf8e..33a51be87c 100644 --- a/frontend/src/lib/components/AssignableTagsInner.svelte +++ b/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 @@ {: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)}
Invalid tag
@@ -194,7 +195,7 @@ {:else}
- {#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}
Interpolated tag based on workspace id the job was created in
{/if} - {#if dynamicTag} -
Interpolated tag based on args input of {dynamicTag}
+ {#if dynamicTag?.kind == 'flow_expr'} +
+ Interpolated tag based on the flow value at {dynamicTag.path}, resolved when + the flow step starts +
+ {:else if dynamicTag} +
Interpolated tag based on args input of {dynamicTag.path}
{/if}
{/if} @@ -252,6 +258,17 @@ > based on args input, use 
$args[a.b.c]
where 
a.b.c
is the path to the value in the args object. +
{#if variant !== 'drawer'}
{/if} + For + dynamic tags + based on flow step results, flow input or flow env, use + 
$flow_expr[results.a.b.c]
where + 
a
is the step id, or + 
$flow_expr[flow_input.a.b.c]
and + 
$flow_expr[flow_env.a.b.c]
. {/if}