feat(pipelines): require data uploads before running a pipeline (#9953)

* feat(pipelines): require data uploads before running a pipeline Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(pipelines): require every S3Object filled for data-upload readiness Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(pipelines): Run pipeline defaults schedule-triggered scripts to their schedule args Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * docs: correct why only schedules default their args in Run pipeline Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(pipelines): close cascade double-start race and gate data-upload readiness on full-schema validity Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix: expose ScriptEditor validity via callback, not banned bindable-with-default Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-07 08:02:40 +00:00 · 2026-07-06 10:26:19 +02:00
parent 4b1f2207b8
commit a1c5b7aa3e
7 changed files with 270 additions and 16 deletions
@@ -164,6 +164,11 @@
 		// succeeded.
 		requireValidAssets?: boolean
 		args: Record<string, any>
+		// Emitted with the test-form's full-schema validity whenever it changes, so
+		// a host (the pipeline editor) can gate a data-upload entry's readiness on
+		// whether every required field is filled, not just the S3 file. A callback
+		// rather than a bindable so we don't hit the `$bindable(default)` ban.
+		onIsValidChange?: (isValid: boolean) => void
 		// Custom timeout (in seconds) from the script settings. Forwarded to the
 		// preview run so "Test" honors the same timeout a deployed run would,
 		// instead of silently falling back to the instance default.
@@ -235,6 +240,7 @@
 		customUi = undefined,
 		requireValidAssets = false,
 		args = $bindable(),
+		onIsValidChange,
 		timeout = undefined,
 		selectedTab = $bindable('main'),
 		hasPreprocessor = $bindable(false),
@@ -678,6 +684,10 @@
 	let jobLoader: JobLoader | undefined = $state(undefined)

 	let isValid: boolean = $state(true)
+	// Mirror the test-form validity out to an optional host callback.
+	$effect(() => {
+		onIsValidChange?.(isValid)
+	})
 	let scriptProgress = $state(undefined)

 	let logPanel: LogPanel | undefined = $state(undefined)
@@ -129,6 +129,9 @@
 		// form (with the auto-generated S3 picker) for the given script.
 		// data_upload has no trigger row; it's a UI-first entry point.
 		onOpenDataUpload?: (scriptPath: string) => void
+		// Data-upload entry scripts whose file has been staged in the run form —
+		// their source node renders green (ready) instead of the neutral prompt.
+		readyDataUploadPaths?: ReadonlySet<string>
 		// Script paths the cursor is over in the Activity panel — a thin neutral
 		// ring each. One entry for a single run row; the whole cascade's runs
 		// when hovering a group header. Distinct from `selectedRunPaths`.
@@ -186,6 +189,7 @@
 		onDeleteTrigger,
 		onOpenWebhook,
 		onOpenDataUpload,
+		readyDataUploadPaths,
 		hoveredPaths,
 		selectedRunPaths,
 		panToNodeId,
@@ -703,6 +707,11 @@
 					runnable_unsaved: info.runnable_path
 						? unsavedRunnablePaths.has(info.runnable_path)
 						: false,
+					// data_upload nodes go green once a file is staged for their
+					// target script (see readyDataUploadPaths / page dataUploadArgs).
+					ready: info.runnable_path
+						? (readyDataUploadPaths?.has(info.runnable_path) ?? false)
+						: false,
 					onCreateMissingTrigger,
 					onEditTrigger,
 					onDeleteTrigger,
@@ -245,6 +245,14 @@
 		// preview only — the client orchestrates the chain). Unset on the deployed
 		// pane, whose single run already cascades via the backend dispatcher.
 		onRunCascadeByPath?: (path: string, args: Record<string, any>) => Promise<string | undefined>
+		// Persisted run-form args for the open data-upload entry (its staged
+		// S3Object), so re-opening the node restores the picked file.
+		runFormInitialArgs?: Record<string, any>
+		// Emitted when the run form's args (or validity) change (read-only
+		// PipelineScriptView OR the edit-mode ScriptEditor test form), so the page
+		// can persist a data-upload entry's staged input (drives node readiness).
+		// `isValid` is the full-schema validity, not just "file present".
+		onRunFormArgsChange?: (path: string, args: Record<string, any>, isValid: boolean) => void
 	}
 	let {
 		selection,
@@ -285,7 +293,9 @@
 		onRequestEdit,
 		canRunByPath = false,
 		onRunByPath,
-		onRunCascadeByPath
+		onRunCascadeByPath,
+		runFormInitialArgs,
+		onRunFormArgsChange
 	}: Props = $props()

 	let readOnly = $derived(mode !== 'edit')
@@ -463,6 +473,31 @@
 		script = undefined
 	})

+	// Data-upload capture (edit mode): the ScriptEditor test form binds `args`, so
+	// mirror it up to the page — that stages a data-upload entry's uploaded /
+	// entered input, driving the node's green "ready" state and seeding the
+	// whole-pipeline run. Reset per script in a pre-effect (before the keyed
+	// ScriptEditor remounts) so switching nodes never leaks one node's input into
+	// the next, seeding from the page's persisted staging so re-opening a node
+	// restores its input. Guarded on the path so a staging round-trip
+	// (emit → page → runFormInitialArgs) doesn't re-seed and loop. The read-only
+	// branch uses PipelineScriptView's own onArgsChange instead.
+	let argsSeedPath: string | undefined = undefined
+	$effect.pre(() => {
+		const p = script?.path
+		if (p === argsSeedPath) return
+		argsSeedPath = p
+		args = runFormInitialArgs ? structuredClone($state.snapshot(runFormInitialArgs)) : {}
+	})
+	// Bound out of ScriptEditor's test-form SchemaForm — full-schema validity of
+	// the edit-mode run form, so the page's readiness check sees whether every
+	// required field (not just the S3 file) is satisfied.
+	let runFormIsValid = $state(true)
+	$effect(() => {
+		if (readOnly || !script) return
+		onRunFormArgsChange?.(script.path, $state.snapshot(args), runFormIsValid)
+	})
+
 	// Persist draft edits back to the parent's drafts Map on transitions
 	// (selection change, pane close), not on every keystroke — a per-key
 	// sync triggered drafts → activeDraft → draftScript → re-clone → emit
@@ -1172,6 +1207,8 @@
 							downstreamCount={downstreamSubscribers}
 							{runsRefreshKey}
 							{runsPendingJobId}
+							initialArgs={runFormInitialArgs}
+							onArgsChange={onRunFormArgsChange}
 							onRunCompleted={() => {
 								previewRefreshKey += 1
 								onRunCompleted?.()
@@ -1234,7 +1271,8 @@
 							bind:assets={liveBodyAssets}
 							bind:inferredColumnLineage={liveColumnLineage}
 							{onTestStateChange}
-							{args}
+							bind:args
+							onIsValidChange={(v) => (runFormIsValid = v)}
 						/>
 					</div>
 				</div>
@@ -69,6 +69,9 @@
 		canRunByPath = false,
 		onRunByPath,
 		onRunCascadeByPath,
+		runFormInitialArgs,
+		onRunFormArgsChange,
+		readyDataUploadPaths,
 		resolveLocalScript,
 		localScriptsVersion,
 		selectionProducers = [],
@@ -162,6 +165,12 @@
 		// Run the open script AND its downstream closure (dev preview only; the
 		// deployed pane leaves this unset — its single run cascades via the backend).
 		onRunCascadeByPath?: (path: string, args: Record<string, any>) => Promise<string | undefined>
+		// Persisted run-form args for the currently-open data-upload entry, and a
+		// callback to persist them as they change — see the page's dataUploadArgs.
+		runFormInitialArgs?: Record<string, any>
+		onRunFormArgsChange?: (path: string, args: Record<string, any>, isValid: boolean) => void
+		// Data-upload entry scripts whose staged upload is ready (green node).
+		readyDataUploadPaths?: Set<string>
 		/** Local-dev (`/pipeline_dev`): resolve a node to its working-tree content
 		 * so the details pane skips the (nonexistent) deployed-script fetch. May
 		 * be async (to infer the args schema for the run form). */
@@ -438,6 +447,7 @@
 					{onDeleteTrigger}
 					{onOpenWebhook}
 					{onOpenDataUpload}
+					{readyDataUploadPaths}
 					onselect={onSelect}
 					{onAddScriptForAsset}
 					{onAddPipelineScript}
@@ -488,6 +498,8 @@
 						{canRunByPath}
 						{onRunByPath}
 						{onRunCascadeByPath}
+						{runFormInitialArgs}
+						{onRunFormArgsChange}
 						{resolveLocalScript}
 						{localScriptsVersion}
 						selection={activeDraft ? undefined : editor.selection}
@@ -33,6 +33,15 @@
 		runsRefreshKey?: any
 		runsPendingJobId?: string | undefined
 		onRunCompleted?: () => void
+		// Seed the run form with previously-staged args (e.g. a data-upload entry
+		// whose S3Object was picked earlier and persisted at the page level). Lets
+		// a re-opened node show the file it already has instead of a blank form.
+		initialArgs?: Record<string, any>
+		// Emitted whenever the run form's args (or validity) change, so the page can
+		// persist a data-upload entry's staged input — that drives the node's
+		// "green / ready" state and seeds the whole-pipeline run. `isValid` is the
+		// full-schema validity (all required fields, not just the S3 file).
+		onArgsChange?: (path: string, args: Record<string, any>, isValid: boolean) => void
 	}

 	let {
@@ -44,12 +53,25 @@
 		downstreamCount = 0,
 		runsRefreshKey,
 		runsPendingJobId,
-		onRunCompleted
+		onRunCompleted,
+		initialArgs,
+		onArgsChange
 	}: Props = $props()

-	let args = $state<Record<string, any>>({})
+	// Seed once from the persisted args (see initialArgs). Cloned so the run
+	// form mutates its own copy, not the page's stored snapshot.
+	// svelte-ignore state_referenced_locally
+	let args = $state<Record<string, any>>(
+		initialArgs ? structuredClone($state.snapshot(initialArgs)) : {}
+	)
 	let isValid = $state(true)
 	let running = $state(false)
+	// Push args + validity back up so the page can persist a staged data-upload
+	// entry. Reading the deep snapshot tracks nested changes (e.g. an S3Object's
+	// `s3` field, or items added to a required array).
+	$effect(() => {
+		onArgsChange?.(script.path, $state.snapshot(args), isValid)
+	})
 	// Offer "Run + downstream" only when a cascade dispatch is wired (dev preview)
 	// and the script actually has subscribers to fan out to.
 	let hasCascade = $derived(!!onRunCascade && downstreamCount > 0)
@@ -60,7 +60,7 @@
 	import { Handle, Position } from '@xyflow/svelte'
 	import { NODE } from '$lib/components/graph/util'
 	import { twMerge } from 'tailwind-merge'
-	import { AlertTriangle, EllipsisVertical, Target, Trash2 } from 'lucide-svelte'
+	import { AlertTriangle, CheckCircle2, EllipsisVertical, Target, Trash2 } from 'lucide-svelte'
 	import DropdownV2 from '$lib/components/DropdownV2.svelte'
 	import { stopPropagation, preventDefault } from 'svelte/legacy'
 	import type { Item } from '$lib/utils'
@@ -101,6 +101,9 @@
 			// auto-generated S3 picker lets the user upload + run) instead of
 			// rendering a "missing" placeholder.
 			onOpenDataUpload?: (scriptPath: string) => void
+			// True once a file is staged for this data_upload entry — renders the
+			// node green so the user knows the pipeline can run (see WIN-2129).
+			ready?: boolean
 			// Page-supplied dispatcher to open the matching native trigger
 			// drawer in edit mode for an attached (non-missing) trigger.
 			// `triggerPath` is the trigger row's path (e.g. the mqtt_trigger
@@ -149,6 +152,10 @@
 	// data_upload routes through its own handler — clicking opens the target
 	// script's run form (with the auto-generated S3 picker).
 	let canOpenDataUpload = $derived(isDataUpload && !!data.runnable_path && !!data.onOpenDataUpload)
+	// A staged upload turns the node green (ready to run); before that it stays
+	// on the neutral surface with the "upload a file" prompt.
+	let dataUploadReady = $derived(isDataUpload && data.ready === true)
+	let DataUploadIcon = $derived(dataUploadReady ? CheckCircle2 : style.icon)
 	// Schedule + the other native kinds all have dedicated editors. Webhook and
 	// data_upload are excluded — they route through their own open handlers.
 	let canCreate = $derived(
@@ -306,26 +313,47 @@
 	{:else if canOpenDataUpload}
 		<!-- Data upload: UI-first entry point (no trigger row). Clicking the
 		     node opens the target script's run form, where the auto-generated
-		     S3 picker lets the user upload a file and run the pipeline. Styled
-		     like an attached trigger, never the red "missing" state. -->
+		     S3 picker lets the user upload a file and run the pipeline. Goes
+		     green once a file is staged (ready), so "Run pipeline" can proceed;
+		     until then it stays neutral with an "upload a file" prompt. Never
+		     the red "missing" state. -->
 		<button
 			type="button"
 			onclick={handleDataUploadClick}
 			class={twMerge(
 				'flex items-center rounded-md drop-shadow-sm overflow-hidden outline outline-1 w-full text-left',
-				style.bg,
-				data.unsaved ? `opacity-80 ${style.borderUnsaved}` : style.border,
+				dataUploadReady
+					? 'bg-green-50 dark:bg-green-900/30 outline-green-500 dark:outline-green-600'
+					: style.bg,
+				dataUploadReady ? '' : data.unsaved ? `opacity-80 ${style.borderUnsaved}` : style.border,
 				'hover:brightness-95 dark:hover:brightness-110 transition-[filter]'
 			)}
 			style="width: {NODE.width}px; min-height: {NODE.height}px;"
-			title={`Data upload for ${data.runnable_path ?? ''} — click to open the run form and upload a file`}
+			title={dataUploadReady
+				? `Data upload for ${data.runnable_path ?? ''} — a file is staged; the pipeline is ready to run. Click to change it.`
+				: `Data upload for ${data.runnable_path ?? ''} — click to open the run form and upload a file`}
 		>
-			<Icon size={14} class={`shrink-0 ml-2 mr-2 ${style.iconText}`} />
+			<DataUploadIcon
+				size={14}
+				class={`shrink-0 ml-2 mr-2 ${dataUploadReady ? 'text-green-600 dark:text-green-400' : style.iconText}`}
+			/>
 			<div class="flex flex-col min-w-0 flex-1 pr-2 py-0.5 leading-tight">
-				<span class="text-3xs uppercase tracking-wide truncate text-tertiary">
-					{style.label}
+				<span
+					class={twMerge(
+						'text-3xs uppercase tracking-wide truncate',
+						dataUploadReady ? 'text-green-700 dark:text-green-400' : 'text-tertiary'
+					)}
+				>
+					{style.label}{dataUploadReady ? ' · ready' : ''}
+				</span>
+				<span
+					class={twMerge(
+						'text-2xs font-mono truncate',
+						dataUploadReady ? 'text-green-700 dark:text-green-400' : 'text-emphasis'
+					)}
+				>
+					{dataUploadReady ? 'File staged' : 'Upload & run'}
 				</span>
-				<span class="text-2xs font-mono truncate text-emphasis"> Upload & run </span>
 			</div>
 		</button>
 	{:else}
@@ -94,6 +94,7 @@
 	import {
 		JobService,
 		OpenAPI,
+		ScheduleService,
 		ScriptService,
 		type AssetKind,
 		type Script,
@@ -1379,6 +1380,43 @@
 	// Run+downstream affordance against overlapping chains stomping each
 	// other's storage writes.
 	let cascadeRunningRoot = $state<string | undefined>(undefined)
+
+	// Script path → its schedule's configured args, so a manual "Run pipeline"
+	// launches a schedule-triggered script with the same payload a real tick
+	// would (rather than empty args). Schedule is the only trigger that stores a
+	// static args payload; every other trigger (webhook / http / event kinds /
+	// data_upload) receives its input from the live event or request at dispatch
+	// time — a message, an HTTP body, an uploaded file — which doesn't exist
+	// during a manual run, so there's nothing to default and those scripts run
+	// empty (data_upload being fed its staged file instead, see dataUploadArgs).
+	// Resolved on demand right before a run (schedule args aren't in the graph
+	// response); fail-safe — a schedule we can't fetch just contributes nothing.
+	let scheduleArgsByPath: Record<string, Record<string, any>> = {}
+	async function resolveScheduleArgs(scripts: string[]): Promise<void> {
+		scheduleArgsByPath = {}
+		const ws = $workspaceStore
+		if (!ws) return
+		const runSet = new Set(scripts)
+		// First schedule (with args) per in-run script; dedupe schedule fetches.
+		const wanted = new Map<string, string>() // script path → schedule path
+		for (const t of displayGraph.triggers) {
+			if (t.trigger_kind !== 'schedule' || t.runnable_kind !== 'script') continue
+			if (!runSet.has(t.runnable_path) || wanted.has(t.runnable_path)) continue
+			if ((t as any).path) wanted.set(t.runnable_path, (t as any).path)
+		}
+		await Promise.all(
+			[...wanted].map(async ([scriptPath, schedulePath]) => {
+				try {
+					const sched = await ScheduleService.getSchedule({ workspace: ws, path: schedulePath })
+					if (sched?.args && Object.keys(sched.args).length > 0) {
+						scheduleArgsByPath[scriptPath] = sched.args as Record<string, any>
+					}
+				} catch {
+					// schedule unreadable / deleted — leave its script on empty args
+				}
+			})
+		)
+	}
 	// Launch one pipeline script for the dev-run cascade. Always passes
 	// _wmill_skip_asset_dispatch — the orchestrator owns the whole closure,
 	// so the backend dispatcher must not double-fire the deployed part of a
@@ -1391,6 +1429,12 @@
 		// drafts (same condition as `displayGraph`). Otherwise — View mode with
 		// drafts hidden — a bounded run must execute the *deployed* scripts the
 		// user is looking at, not preview jobs from hidden local drafts.
+		// Default a script's run args to its schedule's stored payload (the only
+		// trigger with static args — see resolveScheduleArgs), then let a
+		// data-upload entry's staged file override. runWholePipeline already
+		// refused to start unless every data-upload entry is staged, so this is
+		// always the intended input.
+		const staged = { ...(scheduleArgsByPath[path] ?? {}), ...(dataUploadArgs[path]?.args ?? {}) }
 		const draft = mode === 'edit' || includeDrafts ? pe.drafts.get(path) : undefined
 		if (draft) {
 			if (!draft.script.content || !draft.script.language) {
@@ -1402,14 +1446,14 @@
 					content: draft.script.content,
 					language: draft.script.language,
 					path,
-					args: { _wmill_skip_asset_dispatch: true }
+					args: { ...staged, _wmill_skip_asset_dispatch: true }
 				}
 			})
 		}
 		return await JobService.runScriptByPath({
 			workspace: $workspaceStore,
 			path,
-			requestBody: { _wmill_skip_asset_dispatch: true }
+			requestBody: { ...staged, _wmill_skip_asset_dispatch: true }
 		})
 	}
 	// Poll a launched cascade job to a terminal state. Modest fixed cadence —
@@ -1456,9 +1500,13 @@
 				true
 			)
 		}
+		// Claim the running-guard BEFORE the first await so a rapid second click
+		// (which reads `cascadeRunningRoot`) can't slip through and double-launch.
 		cascadeRunningRoot = rootPath
 		let rootJobId: string | undefined
 		try {
+			// Seed schedule-triggered members with their configured payload.
+			await resolveScheduleArgs([rootPath, ...closure.nodes])
 			const res = await runCascade({
 				closure,
 				root: rootPath,
@@ -1637,9 +1685,13 @@
 			sendUserToast('No runnable scripts in this selection', true)
 			return
 		}
+		// Claim the running-guard BEFORE the first await so a rapid second click
+		// (which reads `cascadeRunningRoot`) can't slip through and double-launch.
 		cascadeRunningRoot = schedule.roots[0] ?? scripts[0]
 		let firstJobId: string | undefined
 		try {
+			// Seed schedule-triggered roots with their configured payload.
+			await resolveScheduleArgs(schedule.nodes)
 			const res = await runSelection({
 				schedule,
 				launch: async (path) => {
@@ -1679,6 +1731,71 @@
 		}
 	}

+	// Staged run-form input for data-upload entry scripts, keyed by path. Lifted
+	// out of the (transient, per-selection) run form so the uploaded/entered data
+	// persists across selection changes — it drives each entry node's green
+	// "ready" state and seeds the whole-pipeline run with that input. `valid` is
+	// the run form's full-schema validity (all required fields satisfied).
+	let dataUploadArgs = $state<Record<string, { args: Record<string, any>; valid: boolean }>>({})
+
+	// An S3Object-shaped value (the file picker writes `{ s3: '<path>' }`).
+	function isS3Object(v: any): boolean {
+		return !!v && typeof v === 'object' && !Array.isArray(v) && 's3' in v
+	}
+	// Whether a staged value carries actual data. Covers the two shapes a
+	// data-upload entry takes: an S3Object (file picker → `{ s3: '<path>' }`) and
+	// a plain required input (e.g. a JSON array pasted into the run form).
+	function hasMeaningfulValue(v: any): boolean {
+		if (v == null) return false
+		if (typeof v === 'string') return v.length > 0
+		if (Array.isArray(v)) return v.length > 0
+		if (typeof v === 'object')
+			return isS3Object(v) ? typeof v.s3 === 'string' && v.s3.length > 0 : Object.keys(v).length > 0
+		return true // numbers / booleans count as provided
+	}
+	// A data-upload entry is ready once the user actually provided its data, not
+	// just opened the form. Two guards: the form's own full-schema `valid` (every
+	// required field — including non-file ones — is satisfied), AND that any
+	// declared S3Object file field actually carries a file (the picker can leave
+	// an empty `{ s3: '' }` on a non-required file field, which `valid` alone
+	// wouldn't catch).
+	function dataUploadReady(path: string): boolean {
+		const staged = dataUploadArgs[path]
+		if (!staged || !staged.valid) return false
+		const values = Object.values(staged.args)
+		const s3s = values.filter(isS3Object)
+		if (s3s.length > 0) return s3s.every(hasMeaningfulValue)
+		return values.some(hasMeaningfulValue)
+	}
+	// Persist the run form's args + validity, but only for data-upload entries —
+	// other run forms (partitioned producers) run their own way and must not be
+	// mistaken for a staged upload. Idempotent: the run form re-emits on every
+	// keystroke/validation pass, so bail when the value is unchanged — otherwise
+	// each emit would reassign `dataUploadArgs`, giving `readyDataUploadPaths` a
+	// fresh Set identity that re-syncs the canvas and re-fires the form's emit
+	// effect (effect_update_depth_exceeded).
+	function stageRunFormArgs(path: string, args: Record<string, any>, valid: boolean) {
+		if (!dataUploadEntryPaths.has(path)) return
+		const prev = dataUploadArgs[path]
+		if (prev && prev.valid === valid && JSON.stringify(prev.args) === JSON.stringify(args)) return
+		dataUploadArgs = { ...dataUploadArgs, [path]: { args, valid } }
+	}
+	// Pipeline scripts that are data-upload entry points (a `data_upload` trigger
+	// in the displayed graph). They can't auto-run — they need an uploaded file
+	// before the pipeline can go (see runWholePipeline's gate + the node's green
+	// state).
+	let dataUploadEntryPaths = $derived(
+		new Set(
+			displayGraph.triggers
+				.filter((t) => t.trigger_kind === 'data_upload' && t.runnable_kind === 'script')
+				.map((t) => t.runnable_path)
+		)
+	)
+	// Of those, the ones with a staged file — drives the green node treatment.
+	let readyDataUploadPaths = $derived(
+		new Set([...dataUploadEntryPaths].filter((p) => dataUploadReady(p)))
+	)
+
 	// Every pipeline-member script, for the always-visible header "Run pipeline"
 	// control. Per-node runs are hover/select-gated on the canvas; this
 	// pipeline-level affordance runs the whole graph without hunting for a root
@@ -1695,6 +1812,21 @@
 	// downstream from every source at once. Reuses the same per-hop launch/poll
 	// and one-cascade-at-a-time guard as the node-level chain runs.
 	async function runWholePipeline() {
+		// Data-upload entries can't auto-run with empty args (they'd run against a
+		// missing S3Object). Require every one to be staged (green) first, and
+		// point the user at the first unready node instead of launching a doomed
+		// run.
+		const unready = allPipelineScripts.filter(
+			(p) => dataUploadEntryPaths.has(p) && !dataUploadReady(p)
+		)
+		if (unready.length > 0) {
+			sendUserToast(
+				`Upload data to ${unready.length} data-upload node${unready.length === 1 ? '' : 's'} first — they must be green before the pipeline can run`,
+				true
+			)
+			openDataUploadRun(unready[0])
+			return
+		}
 		await runBoundedCascade(allPipelineScripts)
 	}

@@ -2328,6 +2460,9 @@
 				onDeleteTrigger={mode === 'edit' ? deleteAttachedTrigger : undefined}
 				onOpenWebhook={openWebhookDrawer}
 				onOpenDataUpload={openDataUploadRun}
+				{readyDataUploadPaths}
+				runFormInitialArgs={openScriptPath ? dataUploadArgs[openScriptPath]?.args : undefined}
+				onRunFormArgsChange={stageRunFormArgs}
 				onSelect={handleCanvasSelect}
 				onAddScriptForAsset={mode === 'edit' ? handleAddScriptForAsset : undefined}
 				onAddPipelineScript={mode === 'edit' ? handleAddPipelineScript : undefined}