From 8afe3f056c7fc2df98e62ed237c538496cbe6c92 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 22 Sep 2026 02:10:17 +0200 Subject: [PATCH] fix: recover empty-branch answers, keep decision_job on failure Co-Authored-By: Claude Opus 5 (1M context) --- backend/tests/ai_decision.rs | 48 +++++++++++++++++-- backend/windmill-types/src/flow_status.rs | 6 ++- backend/windmill-worker/src/worker_flow.rs | 35 ++++++++------ .../chat/flow/aiAgentProviders.test.ts | 8 +--- .../copilot/chat/flow/aiAgentProviders.ts | 4 +- .../flows/content/ChoiceBranchList.svelte | 4 +- .../flows/flowDiffManager.svelte.ts | 7 ++- .../src/lib/components/flows/flowExplorer.ts | 4 +- frontend/src/lib/components/flows/flowTree.ts | 30 +++++------- 9 files changed, 92 insertions(+), 54 deletions(-) diff --git a/backend/tests/ai_decision.rs b/backend/tests/ai_decision.rs index 35c7c813aa..2eb7fb5eeb 100644 --- a/backend/tests/ai_decision.rs +++ b/backend/tests/ai_decision.rs @@ -1,5 +1,10 @@ //! AI decision steps run end to end against a stand-in for TypeSafe's System One endpoint. +use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, +}; + use axum::{routing::post, Json, Router}; use serde_json::{json, Value}; use sqlx::{Pool, Postgres}; @@ -10,11 +15,15 @@ use windmill_common::{ }; use windmill_test_utils::*; -/// Answers every question as a choice of `refund`, the way TypeSafe shapes an answer. -async fn start_typesafe_stub() -> anyhow::Result { +/// Answers every question as a choice of `refund`, the way TypeSafe shapes an answer, and counts +/// the calls. +async fn start_typesafe_stub() -> anyhow::Result<(u16, Arc)> { + let calls = Arc::new(AtomicUsize::new(0)); + let counted = calls.clone(); let app = Router::new().route( "/v1/systemone", - post(|Json(body): Json| async move { + post(move |Json(body): Json| async move { + counted.fetch_add(1, Ordering::SeqCst); let answers = body["questions"] .as_object() .into_iter() @@ -35,7 +44,7 @@ async fn start_typesafe_stub() -> anyhow::Result { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; let port = listener.local_addr()?.port(); tokio::spawn(async move { axum::serve(listener, app).await }); - Ok(port) + Ok((port, calls)) } /// The decision chooses its branch from its answers, a branch step reads those answers as @@ -48,7 +57,7 @@ async fn test_ai_decision_runs_the_branch_its_answer_picks( initialize_tracing().await; // The stub listens on loopback, which the SSRF check refuses without this. std::env::set_var("ALLOW_PRIVATE_AI_BASE_URLS", "true"); - let stub_port = start_typesafe_stub().await?; + let (stub_port, calls) = start_typesafe_stub().await?; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); @@ -167,5 +176,34 @@ async fn test_ai_decision_runs_the_branch_its_answer_picks( "the branch reading the previous step was not chosen: {:?}", status.modules[1] ); + + // A retry after its branch fails asks the decision again from its own inputs, rather than + // from the failed branch's args. + let flow: FlowValue = serde_json::from_value(json!({ + "modules": [{ + "id": "d", + "retry": {"constant": {"attempts": 1, "seconds": 0}}, + "value": { + "type": "aidecision", + "input_transforms": decision_inputs, + "branches": [{"expr": "true", "modules": [{"id": "boom", "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(){ throw new Error('branch failed') }", + "input_transforms": {} + }}]}] + } + }] + }))?; + let before = calls.load(Ordering::SeqCst); + let job = run_job_in_new_worker_until_complete( + &db, + false, + JobPayload::RawFlow { value: flow, path: None, restarted_from: None }, + port, + ) + .await; + assert!(!job.success, "the failing branch should fail the flow"); + assert_eq!(calls.load(Ordering::SeqCst) - before, 2); Ok(()) } diff --git a/backend/windmill-types/src/flow_status.rs b/backend/windmill-types/src/flow_status.rs index e284e3e0a2..73e4daf4b1 100644 --- a/backend/windmill-types/src/flow_status.rs +++ b/backend/windmill-types/src/flow_status.rs @@ -303,6 +303,8 @@ pub enum FlowStatusModule { agent_actions: Option>, #[serde(skip_serializing_if = "Option::is_none")] agent_actions_success: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + decision_job: Option, }, } @@ -391,6 +393,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { failed_retries: untagged.failed_retries.unwrap_or_default(), agent_actions: untagged.agent_actions, agent_actions_success: untagged.agent_actions_success, + decision_job: untagged.decision_job, }), other => Err(serde::de::Error::unknown_variant( other, @@ -526,7 +529,8 @@ impl FlowStatusModule { pub fn decision_job(&self) -> Option { match self { FlowStatusModule::InProgress { decision_job, .. } - | FlowStatusModule::Success { decision_job, .. } => *decision_job, + | FlowStatusModule::Success { decision_job, .. } + | FlowStatusModule::Failure { decision_job, .. } => *decision_job, _ => None, } } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 275aa46f53..5f2dd67fe7 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1083,6 +1083,7 @@ pub async fn update_flow_status_after_job_completion_internal( failed_retries: vec![], agent_actions: None, agent_actions_success: None, + decision_job: None, } }; let r = sqlx::query_scalar!( @@ -1370,6 +1371,7 @@ pub async fn update_flow_status_after_job_completion_internal( failed_retries: old_status.retry.failed_jobs.clone(), agent_actions: module_status.agent_actions(), agent_actions_success: module_status.agent_actions_success(), + decision_job: module_status.decision_job(), }), ) } @@ -3304,23 +3306,28 @@ async fn push_next_flow_job( if (flow.modules.is_empty() && !step.is_preprocessor_step()) || matches!(status_module, FlowStatusModule::Success { .. }) { + let result = if flow.modules.is_empty() { + to_raw_value(arc_flow_job_args.as_ref()) + } else if let FlowStatusModule::Success { branch_chosen: Some(_), decision_job, .. } = + &status_module + { + match (&last_job_result, decision_job) { + (Some(result), _) => result.as_ref().clone(), + // Re-entered without the in-memory result (the worker restarted after an empty + // branch was chosen): an AI decision's empty branch returns its answers. + (None, Some(decision_job)) => { + completed_job_result(db, &flow_job.workspace_id, decision_job).await? + } + (None, None) => to_raw_value(&json!("{}")), + } + } else { + // it has to be an empty for loop event + serde_json::from_str("[]").unwrap() + }; return Ok(PushNextFlowJob::Done(Some(UpdateFlow { flow: flow_job.id, success: true, - result: if flow.modules.is_empty() { - to_raw_value(arc_flow_job_args.as_ref()) - } else if matches!( - status_module, - FlowStatusModule::Success { branch_chosen: Some(_), .. } - ) { - last_job_result - .as_ref() - .map(|x| x.as_ref().clone()) - .unwrap_or_else(|| to_raw_value(&json!("{}"))) - } else { - // it has to be an empty for loop event - serde_json::from_str("[]").unwrap() - }, + result, stop_early_override: None, w_id: flow_job.workspace_id.clone(), worker_dir: worker_dir.to_string(), diff --git a/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.test.ts b/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.test.ts index 3c5db54813..06a5c57ed3 100644 --- a/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.test.ts +++ b/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.test.ts @@ -67,13 +67,7 @@ describe('validateAiAgentProviders', () => { expect(() => validateAiAgentProviders(anthropicProvider('claude-something-new'), { ...COMPLETE, - options: [ - { - ...ANTHROPIC, - modelsAreLive: false, - models: { ids: ['claude-sonnet-5'], complete: false } - } - ] + options: [{ ...ANTHROPIC, modelsAreLive: false, models: { ids: ['claude-sonnet-5'], complete: false } }] }) ).not.toThrow() }) diff --git a/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.ts b/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.ts index 36ca2b875e..404b20ef47 100644 --- a/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.ts +++ b/frontend/src/lib/components/copilot/chat/flow/aiAgentProviders.ts @@ -182,9 +182,7 @@ export function selectAiAgentProviderCandidates( if (!configuredPaths.has(candidate.resourcePath)) return 2 return candidate.kind === defaultProviderKind ? 0 : 1 } - return candidates.sort( - (a, b) => rank(a) - rank(b) || a.resourcePath.localeCompare(b.resourcePath) - ) + return candidates.sort((a, b) => rank(a) - rank(b) || a.resourcePath.localeCompare(b.resourcePath)) } /** What a set of modules needs from the catalog: `needsCatalog` is false when no AI agent step diff --git a/frontend/src/lib/components/flows/content/ChoiceBranchList.svelte b/frontend/src/lib/components/flows/content/ChoiceBranchList.svelte index 786e91d7e2..d8197b871d 100644 --- a/frontend/src/lib/components/flows/content/ChoiceBranchList.svelte +++ b/frontend/src/lib/components/flows/content/ChoiceBranchList.svelte @@ -10,7 +10,7 @@ reorderBranches as reorderBranchesOp, graphBranchIndex } from '../branchOps' - import { choiceBranches, type BranchChoiceValue } from '../branchChoice' + import { choiceBranches, hasChoiceBranches, type BranchChoiceValue } from '../branchChoice' import Button from '$lib/components/common/button/Button.svelte' import TextInput from '$lib/components/text_input/TextInput.svelte' import { dragHandle, dragHandleZone } from '@windmill-labs/svelte-dnd-action' @@ -139,7 +139,7 @@ {/each} - {#if value.type === 'branchone' || items.length > 0} + {#if hasChoiceBranches(value)}
Default

Runs if none of the above match

diff --git a/frontend/src/lib/components/flows/flowDiffManager.svelte.ts b/frontend/src/lib/components/flows/flowDiffManager.svelte.ts index 700dca4532..a03e95b521 100644 --- a/frontend/src/lib/components/flows/flowDiffManager.svelte.ts +++ b/frontend/src/lib/components/flows/flowDiffManager.svelte.ts @@ -16,7 +16,12 @@ import { DUPLICATE_MODULE_PREFIX, NEW_MODULE_PREFIX } from './flowDiff' -import { findModuleInFlow, findModuleParent, removeFlowModule, replaceFlowModule } from './flowTree' +import { + findModuleInFlow, + findModuleParent, + removeFlowModule, + replaceFlowModule +} from './flowTree' import { refreshStateStore } from '$lib/svelte5Utils.svelte' import type { StateStore } from '$lib/utils' import type DiffDrawer from '../DiffDrawer.svelte' diff --git a/frontend/src/lib/components/flows/flowExplorer.ts b/frontend/src/lib/components/flows/flowExplorer.ts index ed5ae2c0bf..47b064ea27 100644 --- a/frontend/src/lib/components/flows/flowExplorer.ts +++ b/frontend/src/lib/components/flows/flowExplorer.ts @@ -52,7 +52,7 @@ function filterDependentComponents(modules: FlowModule[], id: string): Record [mod.id, getModuleExprs(mod).filter((expr) => expr.includes(`flow_input`))]) .filter((x) => x[1].length > 0) - ) + ) : Object.fromEntries( modules .map((mod) => [ @@ -63,7 +63,7 @@ function filterDependentComponents(modules: FlowModule[], id: string): Record x[1].length > 0) - ) + ) } function getModuleExprs(x: FlowModule): string[] { diff --git a/frontend/src/lib/components/flows/flowTree.ts b/frontend/src/lib/components/flows/flowTree.ts index 4d832a011b..4eb0691957 100644 --- a/frontend/src/lib/components/flows/flowTree.ts +++ b/frontend/src/lib/components/flows/flowTree.ts @@ -65,7 +65,9 @@ export function getChildModuleBranches(module: FlowModule): FlowModule[][] { } if (module.value.type === 'aiagent' && module.value.tools) { - return [(module.value.tools as FlowModule[]).filter((tool) => isFlowModuleToolCandidate(tool))] + return [ + (module.value.tools as FlowModule[]).filter((tool) => isFlowModuleToolCandidate(tool)) + ] } return [] @@ -270,24 +272,17 @@ export function collectFlowNodes(flow: FlowModuleTree): FlowNodeLocation[] { }) } - visitFlowNodesInModules( - flow.modules ?? [], - (index) => ({ type: 'root', index }), - (match) => { - matches.push({ - module: match.module, - location: match.location - }) - } - ) + visitFlowNodesInModules(flow.modules ?? [], (index) => ({ type: 'root', index }), (match) => { + matches.push({ + module: match.module, + location: match.location + }) + }) return matches } -export function findModuleInModules( - modules: FlowModule[], - moduleId: string -): FlowModule | undefined { +export function findModuleInModules(modules: FlowModule[], moduleId: string): FlowModule | undefined { return ( findFlowNodeInModules(modules, moduleId, (index) => ({ type: 'root', index }))?.module ?? undefined @@ -320,10 +315,7 @@ export function findModuleInFlow(flow: FlowModuleTree, moduleId: string): FlowMo return findFlowNode(flow, moduleId)?.module ?? null } -export function findModuleParent( - flow: FlowModuleTree, - moduleId: string -): ModuleParentLocation | null { +export function findModuleParent(flow: FlowModuleTree, moduleId: string): ModuleParentLocation | null { return findFlowNode(flow, moduleId)?.location ?? null }