fix(frontend): flow editor improvements (#4008)

* fix(frontend): wip

* fix(frontend): wip

* fix(frontend): add missing iter input for while loops

* fix(frontend): improve code

* fix(frontend): improve code

* fix(frontend): failure status for fake modules on hold

* fix(frontend): wip

* fix(frontend): improve flow step warnings

* fix(frontend): add debounce

* fix(frontend): improve rpath detection

* feat(frontend): improve code

* fix(frontend): use the code parser to detect missing dependencies

* fix(frontend): improve code

* fix(frontend): improve code

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -216,6 +216,19 @@
 						true
 					)
 				}
+
+				/**
+				 * else if (mod.type === 'Failure' || mod.type === 'WaitingForPriorSteps') {
+					if (job?.type === 'CompletedJob') {
+						setModuleState('b', {
+							type: 'Failure',
+							args: job?.args,
+							job_id: job?.id,
+							result: job?.result
+						})
+					}
+				}
+				*/
 			})
 		}
 	}

frontend/src/lib/components/flows/content/FlowEditorPanel.svelte

@@ -8,7 +8,7 @@
 	import FlowFailureModule from './FlowFailureModule.svelte'
 	import FlowConstants from './FlowConstants.svelte'
 	import type { FlowModule } from '$lib/gen'
-	import { initRequiredInputFilled } from '../utils'
+	import { initFlowStepWarnings } from '../utils'
 
 	export let noEditor = false
 	export let enableAi = false
@@ -27,10 +27,10 @@
 		}
 	}
 
-	onMount(() => {
-		$flowStore?.value?.modules?.forEach((module) => {
+	async function initWarnings() {
+		for (const module of $flowStore?.value?.modules) {
 			if (!module) {
-				return
+				continue
 			}
 
 			if (!$flowInputsStore) {
@@ -38,12 +38,17 @@
 			}
 
 			$flowInputsStore[module?.id] = {
-				requiredInputsFilled: initRequiredInputFilled(
+				flowStepWarnings: await initFlowStepWarnings(
 					module.value,
-					$flowStateStore?.[module?.id]?.schema ?? {}
+					$flowStateStore?.[module?.id]?.schema ?? {},
+					$flowStore?.value?.modules?.map((m) => m.id) ?? []
 				)
 			}
-		})
+		}
+	}
+
+	onMount(() => {
+		initWarnings()
 	})
 </script>
 

frontend/src/lib/components/flows/content/FlowModuleComponent.svelte

@@ -44,7 +44,8 @@
 	import { enterpriseLicense } from '$lib/stores'
 	import { isCloudHosted } from '$lib/cloud'
 	import { loadSchemaFromModule } from '../flowInfers'
-	import { initRequiredInputFilled, setRequiredInputFilled } from '../utils'
+	import { computeFlowStepWarning, initFlowStepWarnings } from '../utils'
+	import { debounce } from '$lib/utils'
 
 	const {
 		selectedId,
@@ -128,8 +129,12 @@
 			if (inputTransformSchemaForm) {
 				inputTransformSchemaForm.setArgs(input_transforms)
 				if (!deepEqual(schema, $flowStateStore[flowModule.id]?.schema)) {
-					$flowInputsStore![flowModule?.id] = {
-						requiredInputsFilled: initRequiredInputFilled(flowModule.value, schema ?? {})
+					$flowInputsStore[flowModule?.id] = {
+						flowStepWarnings: await initFlowStepWarnings(
+							flowModule.value,
+							schema ?? {},
+							$flowStore?.value?.modules?.map((m) => m?.id) ?? []
+						)
 					}
 				}
 			} else {
@@ -176,21 +181,25 @@
 	let editorPanelSize = noEditor ? 0 : flowModule.value.type == 'script' ? 30 : 50
 	let editorSettingsPanelSize = 100 - editorPanelSize
 
-	function setFlowInput(argName: string) {
-		if ($flowInputsStore && $flowInputsStore?.[flowModule.id] === undefined) {
-			$flowInputsStore[flowModule.id] = {}
-		}
-
+	let debouncedWarning = debounce((argName: string) => {
 		if ($flowInputsStore) {
-			const requiredInputsFilled = setRequiredInputFilled(
+			computeFlowStepWarning(
 				argName,
 				flowModule.value,
-				$flowInputsStore[flowModule.id].requiredInputsFilled ?? {},
-				$flowStateStore[$selectedId]?.schema
-			)
-
-			$flowInputsStore[flowModule.id].requiredInputsFilled = requiredInputsFilled
+				$flowInputsStore[flowModule.id].flowStepWarnings ?? {},
+				$flowStateStore[$selectedId]?.schema ?? {},
+				$flowStore?.value?.modules?.map((m) => m?.id) ?? []
+			).then((flowStepWarnings) => {
+				$flowInputsStore[flowModule.id].flowStepWarnings = flowStepWarnings
+			})
 		}
+	}, 100)
+
+	function setFlowInput(argName: string) {
+		if ($flowInputsStore && flowModule.id && $flowInputsStore?.[flowModule.id] === undefined) {
+			$flowInputsStore[flowModule.id] = {}
+		}
+		debouncedWarning(argName)
 	}
 </script>
 

frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte

@@ -18,13 +18,13 @@
 	import FlowBranchesAllWrapper from './FlowBranchesAllWrapper.svelte'
 	import FlowBranchesOneWrapper from './FlowBranchesOneWrapper.svelte'
 	import FlowWhileLoop from './FlowWhileLoop.svelte'
-	import { initRequiredInputFilled } from '../utils'
+	import { initFlowStepWarnings } from '../utils'
 
 	export let flowModule: FlowModule
 	export let noEditor: boolean = false
 	export let enableAi = false
 
-	const { selectedId, schedule, flowStateStore, flowInputsStore } =
+	const { selectedId, schedule, flowStateStore, flowInputsStore, flowStore } =
 		getContext<FlowEditorContext>('FlowEditorContext')
 
 	let scriptKind: 'script' | 'trigger' | 'approval' = 'script'
@@ -65,9 +65,10 @@
 
 		if ($flowInputsStore) {
 			$flowInputsStore[module?.id] = {
-				requiredInputsFilled: initRequiredInputFilled(
+				flowStepWarnings: await initFlowStepWarnings(
 					module?.value,
-					$flowStateStore[module?.id]?.schema
+					$flowStateStore[module?.id]?.schema,
+					$flowStore.value.modules.map((m) => m.id)
 				)
 			}
 		}
@@ -147,9 +148,10 @@
 
 					if ($flowInputsStore) {
 						$flowInputsStore[module.id] = {
-							requiredInputsFilled: initRequiredInputFilled(
+							flowStepWarnings: await initFlowStepWarnings(
 								module.value,
-								$flowStateStore[module.id].schema
+								$flowStateStore[module.id].schema,
+								$flowStore.value.modules.map((m) => m.id)
 							)
 						}
 					}

frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte

@@ -19,6 +19,7 @@
 	import { fade } from 'svelte/transition'
 	import type { FlowInput } from '../types'
 	import type { Writable } from 'svelte/store'
+	import { twMerge } from 'tailwind-merge'
 
 	export let selected: boolean = false
 	export let deletable: boolean = false
@@ -175,14 +176,30 @@ hover:border-blue-700 {selected ? '' : '!hidden'}"
 			<Move class="mx-[3px]" size={14} strokeWidth={2} />
 		</button>
 
-		{#if (id && !Object.values($flowInputsStore?.[id]?.requiredInputsFilled || {}).every(Boolean)) || Boolean(warningMessage)}
+		{#if (id && Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}).length > 0) || Boolean(warningMessage)}
 			<div class="absolute -top-[10px] -left-[10px]">
 				<Popover>
-					<svelte:fragment slot="text"
-						>{warningMessage ?? 'At least one required input is not set.'}
+					<svelte:fragment slot="text">
+						<ul class="list-disc px-2">
+							{#if id}
+								{#each Object.values($flowInputsStore?.[id]?.flowStepWarnings || {}) as m}
+									<li>
+										{m.message}
+									</li>
+								{/each}
+							{/if}
+						</ul>
 					</svelte:fragment>
 					<div
-						class="flex items-center justify-center h-full w-full rounded-md p-0.5 border border-yellow-600 text-yellow-600 bg-yellow-100 duration-150 hover:bg-yellow-300"
+						class={twMerge(
+							'flex items-center justify-center h-full w-full rounded-md p-0.5 border  duration-150 ',
+							id &&
+								Object.values($flowInputsStore?.[id]?.flowStepWarnings || {})?.some(
+									(x) => x.type === 'error'
+								)
+								? 'border-red-600 text-red-600 bg-red-100 hover:bg-red-300'
+								: 'border-yellow-600 text-yellow-600 bg-yellow-100 hover:bg-yellow-300'
+						)}
 					>
 						<AlertTriangle size={14} strokeWidth={2} />
 					</div>

frontend/src/lib/components/flows/map/FlowModuleSchemaMap.svelte

@@ -11,7 +11,7 @@
 		pickScript
 	} from '$lib/components/flows/flowStateUtils'
 	import type { FlowModule } from '$lib/gen'
-	import { emptyFlowModuleState } from '../utils'
+	import { emptyFlowModuleState, initFlowStepWarnings } from '../utils'
 	import FlowSettingsItem from './FlowSettingsItem.svelte'
 	import FlowConstantsItem from './FlowConstantsItem.svelte'
 
@@ -175,6 +175,30 @@
 	}
 
 	const dispatch = createEventDispatcher()
+
+	async function updateFlowInputsStore() {
+		const keys = Object.keys(dependents ?? {})
+
+		for (const key of keys) {
+			const module = $flowStore.value.modules.find((m) => m.id === key)
+
+			if (!module) {
+				continue
+			}
+
+			if (!$flowInputsStore) {
+				$flowInputsStore = {}
+			}
+
+			$flowInputsStore[module.id] = {
+				flowStepWarnings: await initFlowStepWarnings(
+					module.value,
+					$flowStateStore?.[module.id]?.schema ?? {},
+					$flowStore?.value?.modules?.map((m) => m.id) ?? []
+				)
+			}
+		}
+	}
 </script>
 
 <Portal>
@@ -246,6 +270,8 @@
 						delete $flowInputsStore[e.id]
 					}
 					$flowStore = $flowStore
+
+					updateFlowInputsStore()
 				}
 
 				if (Object.keys(dependents).length > 0) {

frontend/src/lib/components/flows/previousResults.ts

@@ -22,6 +22,14 @@ export function dfs(
 	id: string | undefined,
 	flow: OpenFlow,
 	getParents: boolean = true
+): FlowModule[] {
+	return dfsByModule(id, flow.value.modules, getParents)
+}
+
+export function dfsByModule(
+	id: string | undefined,
+	modules: FlowModule[],
+	getParents: boolean = true
 ): FlowModule[] {
 	if (id === undefined) {
 		return []
@@ -48,7 +56,7 @@ export function dfs(
 		return undefined
 	}
 
-	return rec(id, [flow.value.modules]) ?? []
+	return rec(id, [modules]) ?? []
 }
 
 function getFlowInput(

frontend/src/lib/components/flows/types.ts

@@ -8,7 +8,13 @@ import type { Schedule } from './scheduleUtils'
 export type FlowInput = Record<
 	string,
 	{
-		requiredInputsFilled?: Record<string, boolean>
+		flowStepWarnings?: Record<
+			string,
+			{
+				message: string
+				type: 'error' | 'warning'
+			}
+		>
 	}
 >
 

frontend/src/lib/components/flows/utils.ts

@@ -16,6 +16,7 @@ import type { PickableProperties } from './previousResults'
 import { NEVER_TESTED_THIS_FAR } from './models'
 import { sendUserToast } from '$lib/toast'
 import type { Schema } from '$lib/common'
+import { parseOutputs } from '$lib/infer'
 
 function create_context_function_template(eval_string: string, context: Record<string, any>) {
 	return `
@@ -189,34 +190,102 @@ export function isInputFilled(
 	return true
 }
 
-export function setRequiredInputFilled(
+async function isConnectedToMissingModule(
 	argName: string,
 	flowModuleValue: FlowModuleValue,
-	requiredInputsFilled: Record<string, boolean>,
-	schema: Schema
-) {
-	const type = flowModuleValue.type
-	if (type == 'rawscript' || type == 'script' || type == 'flow') {
-		requiredInputsFilled[argName] = isInputFilled(
-			flowModuleValue.input_transforms,
-			argName,
-			schema ?? {}
-		)
-	}
-
-	return requiredInputsFilled
-}
-
-export function initRequiredInputFilled(flowModuleValue: FlowModuleValue, schema: Schema) {
-	const requiredInputsFilled: Record<string, boolean> = {}
+	moduleIds: string[]
+): Promise<string | undefined> {
 	const type = flowModuleValue.type
 
-	if (type == 'rawscript' || type == 'script' || type == 'flow') {
-		const keys = Object.keys(flowModuleValue.input_transforms)
-		for (const key of keys) {
-			requiredInputsFilled[key] = isInputFilled(flowModuleValue.input_transforms, key, schema ?? {})
+	if (type === 'rawscript' || type === 'script' || type === 'flow') {
+		const input = flowModuleValue?.input_transforms[argName]
+		const val: string = input.type === 'static' ? String(input.value) : input.expr
+
+		try {
+			const outputs = await parseOutputs(val, true)
+			let error: string = ''
+
+			outputs?.forEach(([componentId, id]) => {
+				if (componentId === 'results') {
+					if (!moduleIds.includes(id)) {
+						error += `Input ${argName} is connected to a missing module with id ${id}\n`
+					}
+				}
+			})
+
+			return error
+		} catch (e) {
+			return `Input ${argName} expression is invalid`
 		}
 	}
 
-	return requiredInputsFilled
+	return
+}
+
+export async function computeFlowStepWarning(
+	argName: string,
+	flowModuleValue: FlowModuleValue,
+	messages: Record<
+		string,
+		{
+			message: string
+			type: 'error' | 'warning'
+		}
+	>,
+	schema: Schema,
+	moduleIds: string[] = []
+) {
+	if (messages[argName]) {
+		delete messages[argName]
+	}
+
+	const type = flowModuleValue.type
+	if (type == 'rawscript' || type == 'script' || type == 'flow') {
+		if (!isInputFilled(flowModuleValue.input_transforms, argName, schema ?? {})) {
+			messages[argName] = {
+				message: `Input ${argName} is required but not filled`,
+				type: 'warning'
+			}
+		}
+
+		const errorMessage = await isConnectedToMissingModule(argName, flowModuleValue, moduleIds)
+
+		if (errorMessage) {
+			messages[argName] = {
+				message: errorMessage,
+				type: 'error'
+			}
+		} else {
+			if (messages[argName]?.type === 'error') {
+				delete messages[argName]
+			}
+		}
+	}
+
+	return messages
+}
+
+export async function initFlowStepWarnings(
+	flowModuleValue: FlowModuleValue,
+	schema: Schema,
+	moduleIds: string[] = []
+) {
+	const messages: Record<
+		string,
+		{
+			message: string
+			type: 'error' | 'warning'
+		}
+	> = {}
+	const type = flowModuleValue.type
+
+	if (type == 'rawscript' || type == 'script' || type == 'flow') {
+		const keys = Object.keys(flowModuleValue.input_transforms ?? {})
+		const promises = keys.map(async (key) => {
+			await computeFlowStepWarning(key, flowModuleValue, messages, schema ?? {}, moduleIds)
+		})
+		await Promise.all(promises)
+	}
+
+	return messages
 }

frontend/src/lib/components/graph/FlowGraph.svelte

@@ -26,6 +26,7 @@
 	import { deepEqual } from 'fast-equals'
 	import DarkModeObserver from '../DarkModeObserver.svelte'
 	import type { FlowInput } from '../flows/types'
+	import { dfsByModule } from '../flows/previousResults'
 
 	export let success: boolean | undefined = undefined
 	export let modules: FlowModule[] | undefined = []
@@ -184,21 +185,35 @@
 
 			if (useDataflow && $selectedId) {
 				let deps = getDependeeAndDependentComponents($selectedId, modules ?? [], failureModule)
+
 				if (deps) {
 					Object.entries(deps.dependees).forEach((x, i) => {
-						let pid = x[0]
-						edges.push({
-							id: `dep-${pid}-${$selectedId}`,
-							source: pid,
-							target: $selectedId!,
-							labelBgColor: darkMode ? '#999' : 'white',
-							edgeColor: darkMode ? 'white' : 'black',
-							arrow: false,
-							animate: true,
-							noHandle: true,
-							label: pid,
-							type: 'bezier',
-							offset: i * 20
+						const inputs = x[1]
+
+						inputs?.forEach((input, index) => {
+							let pid = x[0]
+
+							if (input?.startsWith('flow_input.iter')) {
+								const parent = dfsByModule($selectedId!, modules ?? [])?.pop()
+
+								if (parent?.id) {
+									pid = parent.id
+								}
+							}
+
+							edges.push({
+								id: `dep-${pid}-${$selectedId}`,
+								source: pid,
+								target: $selectedId!,
+								labelBgColor: darkMode ? '#999' : 'white',
+								edgeColor: darkMode ? 'white' : 'black',
+								arrow: false,
+								animate: true,
+								noHandle: true,
+								label: pid,
+								type: 'bezier',
+								offset: index * 20
+							})
 						})
 					})