diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index 50b5c8da95..ac941d0bf0 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -164,6 +164,11 @@ // succeeded. requireValidAssets?: boolean args: Record + // 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) diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 6f8460fa10..fbaf3ed6ed 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -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 // 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, diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 095169e3b7..a097f09b6c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -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) => Promise + // Persisted run-form args for the open data-upload entry (its staged + // S3Object), so re-opening the node restores the picked file. + runFormInitialArgs?: Record + // 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, 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)} /> diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 6e6a694e82..cc44d0b41a 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -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) => Promise + // 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 + onRunFormArgsChange?: (path: string, args: Record, isValid: boolean) => void + // Data-upload entry scripts whose staged upload is ready (green node). + readyDataUploadPaths?: Set /** 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} diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineScriptView.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineScriptView.svelte index 18564c6edd..79a1cf6d47 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineScriptView.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineScriptView.svelte @@ -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 + // 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, isValid: boolean) => void } let { @@ -44,12 +53,25 @@ downstreamCount = 0, runsRefreshKey, runsPendingJobId, - onRunCompleted + onRunCompleted, + initialArgs, + onArgsChange }: Props = $props() - let args = $state>({}) + // 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>( + 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) diff --git a/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte b/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte index b888ae226a..98307013c4 100644 --- a/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte @@ -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} + 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. --> {:else} diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index e65c97b83d..ad2f63e388 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -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(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> = {} + async function resolveScheduleArgs(scripts: string[]): Promise { + 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() // 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 + } + } 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; valid: boolean }>>({}) + + // An S3Object-shaped value (the file picker writes `{ s3: '' }`). + 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: '' }`) 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, 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}