kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: give access to results in early stop expr (#7514)

Browse Source 
* feat: give access to results in early stop expr

* nit
This commit is contained in:
hugocasa
2026-01-07 14:06:56 +07:00
committed by GitHub
parent b31d8dffc3
commit 4763eda6b7
6 changed files with 71 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
bc904859dd66c55ebad002e8526103c73de841cd
cf96b45aa1183f15b3cc1b971035de5e37a68849

2
backend/windmill-worker/src/ai_executor.rs
View File

@@ -508,7 +508,7 @@ pub async fn run_agent(
.clone()
.unwrap_or_else(|| "unknown".to_string());
Some(get_transform_context(job, &previous_id, flow_status).await?)
Some(get_transform_context(job, &previous_id, flow_status))
} else {
None
}

68
backend/windmill-worker/src/worker_flow.rs
View File

@@ -283,6 +283,24 @@ fn get_stop_after_if_data(stop_after_if: Option<&StopAfterIf>) -> (bool, Option<
return (false, None);
}
async fn get_id_ctx_for_expr(
expr: &str,
flow: uuid::Uuid,
db: &DB,
status: &FlowStatus,
) -> error::Result<Option<IdContext>> {
if expr.contains("results.") || expr.contains("results[") || expr.contains("results?.") {
let flow_job = get_mini_pulled_job(db, &flow).await?;
if let Some(flow_job) = flow_job {
Ok(Some(get_transform_context(&flow_job, "", &status)))
} else {
Ok(None)
}
} else {
Ok(None)
}
}
async fn evaluate_stop_after_all_iters_if(
db: &DB,
stop_after_all_iters_if: &StopAfterIf,
@@ -294,6 +312,8 @@ async fn evaluate_stop_after_all_iters_if(
stop_early_err_msg: &mut Option<String>,
nresult: &mut Option<Arc<Box<RawValue>>>,
args: HashMap<String, Box<RawValue>>,
flow: uuid::Uuid,
status: &FlowStatus,
) -> error::Result<()> {
let iters_result = match &module_status {
FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), .. } => {
@@ -308,13 +328,15 @@ async fn evaluate_stop_after_all_iters_if(
*nresult = Some(iters_result.clone()); // as an optimization, we store the result of all jobs as when stop_early_after_all_iters evaluates to false, it would have to be computed (finished loop/branchall)
let id_ctx = get_id_ctx_for_expr(&stop_after_all_iters_if.expr, flow, db, status).await?;
let stop_early_after_all_iters = compute_bool_from_expr(
&stop_after_all_iters_if.expr,
Marc::new(args),
None,
iters_result.clone(),
None,
None,
id_ctx.as_ref(),
Some(client),
None,
None,
@@ -542,13 +564,15 @@ pub async fn update_flow_status_after_job_completion_internal(
};
let args = from_result_to_args(args.as_ref().await.get_ref())?;
let id_ctx = get_id_ctx_for_expr(expr, flow, db, &old_status).await?;
compute_bool_from_expr(
&expr,
Marc::new(args),
None,
result.clone(),
all_iters,
None,
id_ctx.as_ref(),
Some(client),
None,
None,
@@ -816,6 +840,8 @@ pub async fn update_flow_status_after_job_completion_internal(
&mut stop_early_err_msg,
&mut nresult,
args,
flow,
&old_status,
)
.await?;
}
@@ -1023,6 +1049,8 @@ pub async fn update_flow_status_after_job_completion_internal(
&mut stop_early_err_msg,
&mut nresult,
args,
flow,
&old_status,
)
.await?;
}
@@ -2917,9 +2945,7 @@ async fn push_next_flow_job(
drop(resume_messages);
let is_skipped = if let Some(skip_if) = &module.skip_if {
let idcontext = get_transform_context(&flow_job, previous_id.as_str(), &status)
.warn_after_seconds(3)
.await?;
let idcontext = get_transform_context(&flow_job, previous_id.as_str(), &status);
compute_bool_from_expr(
&skip_if.expr,
arc_flow_job_args.clone(),
@@ -2999,9 +3025,7 @@ async fn push_next_flow_job(
| FlowModuleValue::Flow { input_transforms, .. }
| FlowModuleValue::AIAgent { input_transforms, .. },
) => {
let ctx = get_transform_context(&flow_job, &previous_id, &status)
.warn_after_seconds(3)
.await?;
let ctx = get_transform_context(&flow_job, &previous_id, &status);
transform_context = Some(ctx);
let by_id = transform_context.as_ref().unwrap();
// if a failure step, we add flow job id and started_at to the context. This is for error handling of triggers where we wrap scripts into single step flows
@@ -3203,9 +3227,7 @@ async fn push_next_flow_job(
args.insert("iter".to_string(), to_raw_value(new_args));
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)
.warn_after_seconds(3)
.await?;
let ctx = get_transform_context(&flow_job, "", &status);
let ti = transform_input(
Marc::new(args),
flow.flow_env.as_ref(),
@@ -3260,9 +3282,7 @@ async fn push_next_flow_job(
to_raw_value(&json!({ "index": i as i32, "value": itered[i]})),
);
if let Some(input_transforms) = simple_input_transforms {
let ctx = get_transform_context(&flow_job, &previous_id, &status)
.warn_after_seconds(3)
.await?;
let ctx = get_transform_context(&flow_job, &previous_id, &status);
let ti = transform_input(
Marc::new(hm),
flow.flow_env.as_ref(),
@@ -3377,9 +3397,7 @@ async fn push_next_flow_job(
}
let evaluated_timeout = if let Some(timeout_transform) = &module.timeout {
let ctx = get_transform_context(&flow_job, &previous_id, &status)
.warn_after_seconds(3)
.await?;
let ctx = get_transform_context(&flow_job, &previous_id, &status);
let timeout_value = evaluate_input_transform::<i32>(
timeout_transform,
@@ -3458,9 +3476,7 @@ async fn push_next_flow_job(
if let Some(parallelism_transform) = &value_with_parallel.parallelism {
tracing::debug!(id = %flow_job.id, root_id = %job_root, "evaluating parallelism expression for forloopflow job {uuid}");
let ctx = get_transform_context(&flow_job, &previous_id, &status)
.warn_after_seconds(3)
.await?;
let ctx = get_transform_context(&flow_job, &previous_id, &status);
let evaluated_parallelism = evaluate_input_transform::<u16>(
parallelism_transform,
@@ -4297,7 +4313,7 @@ async fn compute_next_flow_transform(
| FlowStatusModule::WaitingForEvents { .. }
| FlowStatusModule::WaitingForExecutor { .. } => {
let mut branch_chosen = BranchChosen::Default;
let idcontext = get_transform_context(&flow_job, previous_id, &status).await?;
let idcontext = get_transform_context(&flow_job, previous_id, &status);
for (i, b) in branches.iter().enumerate() {
let pred = compute_bool_from_expr(
&b.expr,
@@ -4580,7 +4596,7 @@ async fn next_forloop_status(
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
get_transform_context(&flow_job, previous_id, &status)
};
/* Iterator is an InputTransform, evaluate it into an array. */
let itered_raw = match iterator {
@@ -4654,7 +4670,7 @@ async fn next_forloop_status(
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
get_transform_context(&flow_job, previous_id, &status)
};
let itered_raw = match iterator {
InputTransform::Static { value } => to_raw_value(value),
@@ -4930,18 +4946,18 @@ pub async fn script_to_payload(
})
}
pub async fn get_transform_context(
pub fn get_transform_context(
flow_job: &MiniPulledJob,
previous_id: &str,
status: &FlowStatus,
) -> error::Result<IdContext> {
) -> IdContext {
let steps_results: HashMap<String, JobResult> = status
.modules
.iter()
.filter_map(|x| x.job_result().map(|y| (x.id(), y)))
.collect();
Ok(IdContext { flow_job: flow_job.id, steps_results, previous_id: previous_id.to_string() })
IdContext { flow_job: flow_job.id, steps_results, previous_id: previous_id.to_string() }
}
// trait IntoArray: Sized {

18
frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte
View File

@@ -51,10 +51,10 @@
return null
}
let raise_error_message_stop_after_all_if = $state(
flowModule.stop_after_all_iters_if?.error_message !== undefined
flowModule.stop_after_all_iters_if?.error_message != undefined
)
let raise_error_message_stop_after_if = $state(
flowModule.stop_after_if?.error_message !== undefined
flowModule.stop_after_if?.error_message != undefined
)
let { isLoop, isParallelLoop } = $derived(
flowModule.value.type === 'forloopflow' || flowModule.value.type === 'whileloopflow'
@@ -166,8 +166,7 @@
<PropPickerWrapper
noPadding
notSelectable
flow_input={stepPropPicker.pickableProperties.flow_input}
pickableProperties={undefined}
pickableProperties={stepPropPicker.pickableProperties}
result={earlyStopResult}
extraResults={isLoop ? { all_iters: result } : undefined}
on:select={({ detail }) => {
@@ -180,8 +179,8 @@
lang="javascript"
bind:code={flowModule.stop_after_if.expr}
class="h-full"
extraLib={`declare const result = ${JSON.stringify(earlyStopResult)};` +
`\n declare const flow_input = ${JSON.stringify(stepPropPicker.pickableProperties.flow_input)};` +
extraLib={`declare const result = ${JSON.stringify(earlyStopResult)};\n` +
stepPropPicker.extraLib +
(isLoop ? `\ndeclare const all_iters = ${JSON.stringify(result)};` : '')}
/>
</PropPickerWrapper>
@@ -303,8 +302,7 @@
<PropPickerWrapper
notSelectable
noPadding
flow_input={stepPropPicker.pickableProperties.flow_input}
pickableProperties={undefined}
pickableProperties={stepPropPicker.pickableProperties}
{result}
on:select={({ detail }) => {
editor?.insertAtCursor(detail)
@@ -316,8 +314,8 @@
lang="javascript"
bind:code={flowModule.stop_after_all_iters_if.expr}
class="h-full"
extraLib={`declare const result = ${JSON.stringify(result)};` +
`\ndeclare const flow_input = ${JSON.stringify(stepPropPicker.pickableProperties.flow_input)};`}
extraLib={`declare const result = ${JSON.stringify(result)};\n` +
stepPropPicker.extraLib}
/>
</PropPickerWrapper>
</div>

4
frontend/src/lib/components/flows/propPicker/PropPickerWrapper.svelte
View File

@@ -144,7 +144,7 @@
: 'transparent'}
animationDuration="4s"
>
{#if result != undefined}
{#if result != undefined && !pickableProperties}
<PropPickerResult
{result}
{extraResults}
@@ -159,6 +159,8 @@
/>
{:else if pickableProperties}
<PropPicker
{result}
{extraResults}
{displayContext}
{error}
{flow_env}

18
frontend/src/lib/components/propertyPicker/PropPicker.svelte
View File

@@ -19,6 +19,8 @@
export let allowCopy = false
export let previousId: string | undefined = undefined
export let flow_env: Record<string, any> | undefined = undefined
export let result: any | undefined = undefined
export let extraResults: any = undefined
let variables: Record<string, string> = {}
let resources: Record<string, any> = {}
@@ -69,7 +71,9 @@
: keepByKey(pickableProperties.priorIds, search)
flowEnvFiltered =
search === EMPTY_STRING ? pickableProperties.flow_env : keepByKey(pickableProperties.flow_env, search)
search === EMPTY_STRING
? pickableProperties.flow_env
: keepByKey(pickableProperties.flow_env, search)
}, 50)
}
@@ -221,7 +225,7 @@
await updateCollapsable()
}
$: search, $inputMatches, $propPickerConfig, pickableProperties, updateState()
$: (search, $inputMatches, $propPickerConfig, pickableProperties, updateState())
onDestroy(() => {
clearTimeout(timeout)
@@ -251,6 +255,16 @@
<Badge small>filter: {filteringFlowInputsOrResult}</Badge>
</div>
{/if}
{#if result != undefined}
<span class={categoryTitleClasses}>Step Result</span>
<div class={categoryContentClasses}>
<ObjectViewer
{allowCopy}
json={{ result, ...(extraResults ? extraResults : {}) }}
on:select
/>
</div>
{/if}
{#if flowInputsFiltered && (Object.keys(flowInputsFiltered ?? {}).length > 0 || !filterActive)}
<span class={categoryTitleClasses}>Flow Input</span>
<div class={categoryContentClasses}>