From 9e5982769a2d37a3a9bc1cd1efbe6580be23df93 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 1 Oct 2026 15:11:29 +0200 Subject: [PATCH] feat: data upload opens on its inputs alone, works in sessions, with a prominent Upload data button Co-Authored-By: Claude Opus 5.5 (1M context) --- frontend/src/lib/components/Editor.svelte | 3 + .../src/lib/components/ScriptEditor.svelte | 71 +++++----- .../assets/AssetGraph/AssetGraphCanvas.svelte | 2 + .../AssetGraph/AssetGraphDetailsPane.svelte | 40 +++++- .../assets/AssetGraph/AssetNode.svelte | 28 +++- .../AssetGraph/PipelineGraphEditor.svelte | 17 +++ .../AssetGraph/PipelineScriptView.svelte | 41 ++++-- .../AssetGraph/pipelineDataUploads.svelte.ts | 87 +++++++++++++ .../sessions/PipelineEditorView.svelte | 25 +++- .../(logged)/pipeline/[folder]/+page.svelte | 121 ++++++++---------- 10 files changed, 313 insertions(+), 122 deletions(-) create mode 100644 frontend/src/lib/components/assets/AssetGraph/pipelineDataUploads.svelte.ts diff --git a/frontend/src/lib/components/Editor.svelte b/frontend/src/lib/components/Editor.svelte index 5ec3c62b5e..d771fb67e5 100644 --- a/frontend/src/lib/components/Editor.svelte +++ b/frontend/src/lib/components/Editor.svelte @@ -1523,6 +1523,9 @@ } catch (e) { console.log('error initializing services', e) } + // Unmounted while the services loaded: the container is detached, and + // creating the editor on it throws. + if (destroyed || !divEl?.isConnected) return // vscode.languages.registerDefinitionProvider('*', { // provideDefinition(document, position, token) { diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index e3215aa9cf..40fb5ebead 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -2068,43 +2068,15 @@ {/if} {#if customUi?.previewPanel?.hideArgs} - -
- {#if testJob?.id && testJob.type === 'CompletedJob' && opWs} - -
- - {testJob.id.slice(0, 8)}… ↗ - - -
- {/if} -
+ +
+ +
+
{#if testIsLoading} {@render cancelTestButton('sm', 'shadow-md')} {:else if (customUi?.previewPanel?.downstreamSubscribers ?? 0) > 0 || customUi?.previewPanel?.onBoundedRun} @@ -2238,6 +2210,31 @@ {@render runTestButton('sm', 'shadow-md')} {/if}
+ {#if testJob?.id && testJob.type === 'CompletedJob' && opWs} + +
+ + {testJob.id.slice(0, 8)}… ↗ + + +
+ {/if} +
{#if customUi?.previewPanel?.argsAboveLogs && schema?.properties && Object.keys(schema.properties).length > 0}
, isValid: boolean) => void + // Show only the open script's inputs and runs (a data-upload click), not + // its editor. `onRunInputs` runs it with them, drafts included, and + // `onShowScript` leaves this view for the script. + inputsOnly?: boolean + onRunInputs?: (path: string, args: Record) => Promise + onShowScript?: () => void } let { selection, @@ -351,7 +357,10 @@ onRunByPath, onRunCascadeByPath, runFormInitialArgs, - onRunFormArgsChange + onRunFormArgsChange, + inputsOnly = false, + onRunInputs, + onShowScript }: Props = $props() let readOnly = $derived(mode !== 'edit') @@ -433,7 +442,7 @@ // PipelineScriptView renders instead of the test panel. let viewReadyForTarget = $derived.by( () => - readOnly && + (readOnly || inputsOnly) && runTargetPath !== undefined && script?.path === runTargetPath && (isDraft || (!scriptRes.loading && scriptRes.current?.path === runTargetPath)) @@ -574,7 +583,9 @@ let args = $state>({}) let argsSeedPath: string | undefined = undefined $effect.pre(() => { - const p = script?.path + // Re-seeded on leaving the inputs-only view too, so the file picked there + // shows in the editor's test form. + const p = script?.path === undefined ? undefined : `${script.path}|${inputsOnly}` if (p === argsSeedPath) return argsSeedPath = p args = runFormInitialArgs ? structuredClone($state.snapshot(runFormInitialArgs)) : {} @@ -584,7 +595,8 @@ // required field (not just the S3 file) is satisfied. let runFormIsValid = $state(true) $effect(() => { - if (readOnly || !script) return + // Inputs-only mounts no ScriptEditor: PipelineScriptView reports its own args. + if (readOnly || inputsOnly || !script) return onRunFormArgsChange?.(script.path, $state.snapshot(args), runFormIsValid) }) @@ -1333,6 +1345,26 @@
Failed to load: {scriptRes.error.message}
+ {:else if script && inputsOnly} + {#key script.path} + { + previewRefreshKey += 1 + onRunCompleted?.() + }} + /> + {/key} {:else if script && readOnly} + + e.stopPropagation()} + onkeydown={(e) => e.stopPropagation()} + > + + + {:else if trigger?.onOpen} + {/if} {#if hasCascade}
-
- -
+ {#if !inputsOnly} +
+ +
+ {/if}
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineDataUploads.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineDataUploads.svelte.ts new file mode 100644 index 0000000000..8205b5f903 --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineDataUploads.svelte.ts @@ -0,0 +1,87 @@ +import { untrack } from 'svelte' +import type { AssetGraphResponse } from './types' + +type Staged = { args: Record; 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 +} + +/** + * Staged run-form input for the graph's 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 a run with that input. `valid` is the run form's + * full-schema validity (all required fields satisfied). + */ +export function usePipelineDataUploads(getGraph: () => AssetGraphResponse) { + let staged = $state>({}) + + // Scripts that are data-upload entry points (a `data_upload` trigger in the + // displayed graph). They can't auto-run — they need their data first. + const entryPaths = $derived( + new Set( + getGraph() + .triggers.filter((t) => t.trigger_kind === 'data_upload' && t.runnable_kind === 'script') + .map((t) => t.runnable_path) + ) + ) + + // 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 isReady(path: string): boolean { + const s = staged[path] + if (!s || !s.valid) return false + const values = Object.values(s.args) + const s3s = values.filter(isS3Object) + if (s3s.length > 0) return s3s.every(hasMeaningfulValue) + return values.some(hasMeaningfulValue) + } + + const readyPaths = $derived(new Set([...entryPaths].filter((p) => isReady(p)))) + + return { + get entryPaths(): ReadonlySet { + return entryPaths + }, + /** Entry scripts with their data provided — drives the green node treatment. */ + get readyPaths(): ReadonlySet { + return readyPaths + }, + isReady, + argsFor: (path: string): Record | undefined => staged[path]?.args, + // 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 `staged`, giving `readyPaths` a fresh Set identity + // that re-syncs the canvas and re-fires the form's emit effect + // (effect_update_depth_exceeded). + // Called from the run forms' own effects, so its reads are untracked: a form + // must not re-run because the store it just wrote to changed. + stage: (path: string, args: Record, valid: boolean) => + untrack(() => { + if (!entryPaths.has(path)) return + const prev = staged[path] + if (prev && prev.valid === valid && JSON.stringify(prev.args) === JSON.stringify(args)) + return + staged = { ...staged, [path]: { args, valid } } + }) + } +} diff --git a/frontend/src/lib/components/sessions/PipelineEditorView.svelte b/frontend/src/lib/components/sessions/PipelineEditorView.svelte index a1936791dd..d92f856b41 100644 --- a/frontend/src/lib/components/sessions/PipelineEditorView.svelte +++ b/frontend/src/lib/components/sessions/PipelineEditorView.svelte @@ -4,6 +4,7 @@ import { Loader2, Save, Workflow } from 'lucide-svelte' import PipelineGraphEditor from '$lib/components/assets/AssetGraph/PipelineGraphEditor.svelte' import PipelineTriggerEditors from '$lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte' + import { usePipelineDataUploads } from '$lib/components/assets/AssetGraph/pipelineDataUploads.svelte' import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph' import { useActiveRunnableIds } from '$lib/components/assets/AssetGraph/activeRunnables.svelte' import type { @@ -142,6 +143,7 @@ }) function handleCanvasSelect(s: AssetGraphSelection | undefined) { + uploadInputsPath = undefined if (s && s.kind === 'runnable' && s.runnable_kind === 'script' && pe.drafts.has(s.path)) { pe.activeDraftPath = s.path pe.selection = undefined @@ -202,7 +204,11 @@ // Data upload has no trigger row — open the target in the details pane and pulse // the signal so its auto-generated run form focuses the S3 input. + // A data-upload click opens the node on its inputs alone; any other selection + // (or "Open script") goes back to the editor. + let uploadInputsPath = $state(undefined) function openDataUploadRun(scriptPath: string) { + uploadInputsPath = scriptPath if (pe.drafts.has(scriptPath)) { pe.activeDraftPath = scriptPath pe.selection = undefined @@ -244,12 +250,19 @@ } }) + const dataUploads = usePipelineDataUploads(() => resolvedGraph) + function runProducer(producer: { kind: 'script' | 'flow'; path: string; cascade?: boolean }) { // Pipeline nodes are scripts; a flow producer can't be preview/by-path run. if (producer.kind !== 'script') return Promise.resolve(undefined) - return runNode(producer.path, {}, producer.cascade ?? false) + // A data-upload entry runs on the file staged in its run form. + return runNode(producer.path, dataUploads.argsFor(producer.path) ?? {}, producer.cascade ?? false) } + // Tell the open node's runs list a job started, so it shows it right away. + let runsRefreshKey = $state(0) + let runsPendingJobId = $state(undefined) + async function runNode( nodePath: string, args: Record = {}, @@ -283,6 +296,8 @@ } activeRunnable = { kind: 'script', path: nodePath } activeRunnableJobId = jobId + runsPendingJobId = jobId + runsRefreshKey++ return jobId } catch (e: any) { sendUserToast(`Run failed: ${e?.body ?? e?.message ?? e}`, true) @@ -483,12 +498,20 @@ onOpenWebhook={openWebhookDrawer} onOpenDataUpload={openDataUploadRun} focusUploadSignal={focusDataUploadSignal} + readyDataUploadPaths={dataUploads.readyPaths} + runFormInitialArgs={pe.openScriptPath ? dataUploads.argsFor(pe.openScriptPath) : undefined} + onRunFormArgsChange={dataUploads.stage} {activeRunnable} + {runsRefreshKey} + {runsPendingJobId} activeRunnableIds={activeRunnables.ids} runStates={activeRunnables.states} eventLogEvents={activeRunnables.events} onRunProducer={runProducer} onRunByPath={(path, args) => runNode(path, args)} + inputsOnly={uploadInputsPath !== undefined && uploadInputsPath === pe.openScriptPath} + onRunInputs={(path, args) => runNode(path, args)} + onShowScript={() => (uploadInputsPath = undefined)} canRunByPath onTestStateChange={(running) => { const openPath = pe.openScriptPath diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index bdc2c29554..d4636f05ca 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -45,6 +45,7 @@ } from '$lib/components/assets/AssetGraph/resolveGraph' import { normalizePipelineFolder } from '$lib/utils/pipelineFolder' import { hideDbtRunnables } from '$lib/components/assets/AssetGraph/hideDbtRunnables' + import { usePipelineDataUploads } from '$lib/components/assets/AssetGraph/pipelineDataUploads.svelte' import { buildSchemaContractContext } from '$lib/components/assets/AssetGraph/schemaContracts' import { computeDownstreamClosure, @@ -995,6 +996,7 @@ // template can gate them per-mode with simple ternaries — the canvas // hides each affordance when its callback is undefined. function handleCanvasSelect(s: AssetGraphSelection | undefined) { + uploadInputsPath = undefined // Clicking a node while the pane is explicitly hidden is a request // to see that node — unhide. Background clicks (s == undefined) // keep the hidden state. @@ -1904,7 +1906,7 @@ // 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). + // empty (data_upload being fed its staged file instead, see dataUploads). // 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> = {} @@ -1950,7 +1952,7 @@ // 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 staged = { ...(scheduleArgsByPath[path] ?? {}), ...(dataUploads.argsFor(path) ?? {}) } const draft = mode === 'edit' || includeDrafts ? pe.drafts.get(path) : undefined if (draft) { if (!draft.script.content || !draft.script.language) { @@ -2266,70 +2268,8 @@ } } - // 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))) - ) + // Data-upload entries' staged input; see usePipelineDataUploads. + const dataUploads = usePipelineDataUploads(() => displayGraph) // Every pipeline-member script, for the always-visible header "Run pipeline" // control. Per-node runs are hover/select-gated on the canvas; this @@ -2352,7 +2292,7 @@ // point the user at the first unready node instead of launching a doomed // run. const unready = allPipelineScripts.filter( - (p) => dataUploadEntryPaths.has(p) && !dataUploadReady(p) + (p) => dataUploads.entryPaths.has(p) && !dataUploads.isReady(p) ) if (unready.length > 0) { sendUserToast( @@ -2559,7 +2499,45 @@ // auto-generated run form (the script declares an `S3Object` input, so the // form renders the S3 picker) lets the user upload a file and run the // pipeline. Drafts open in-place too so the user can fill the body first. + // A data-upload click opens the node on its inputs alone; any other selection + // (or "Open script") goes back to the script. + let uploadInputsPath = $state(undefined) + // Runs the open node with its form inputs: a draft previews its own content, a + // deployed script runs for real (in view mode, cascading like any legit run). + async function runUploadInputs( + path: string, + args: Record + ): Promise { + const draft = mode === 'edit' ? pe.drafts.get(path) : undefined + if (!draft) return mode === 'edit' ? runDeployedWithArgs(path, args) : runByPathLegit(path, args) + if (!$workspaceStore || !draft.script.content || !draft.script.language) return undefined + activeRunnables.arm(`script:${path}`) + try { + const jobId = await JobService.runScriptPreview({ + workspace: $workspaceStore, + requestBody: { + content: draft.script.content, + language: draft.script.language, + path, + args: { ...args, _wmill_skip_asset_dispatch: true } + } + }) + runsPendingJobId = jobId + runsRefreshKey++ + activeRunnable = { kind: 'script', path } + activeRunnableJobId = jobId + return jobId + } catch (e: any) { + sendUserToast(`Run failed: ${e?.body ?? e?.message ?? e}`, true) + return undefined + } + } + // Same as a canvas run of a deployed node in edit mode: this step only. + function runDeployedWithArgs(path: string, args: Record) { + return runByPathLegit(path, { ...args, _wmill_skip_asset_dispatch: true }) + } function openDataUploadRun(scriptPath: string) { + uploadInputsPath = scriptPath // Opening the run form is meaningless while the pane is hidden — // unhide first, same as a plain node click. panelHidden = false @@ -3019,9 +2997,12 @@ onDeleteTrigger={mode === 'edit' ? deleteAttachedTrigger : undefined} onOpenWebhook={openWebhookDrawer} onOpenDataUpload={openDataUploadRun} - {readyDataUploadPaths} - runFormInitialArgs={openScriptPath ? dataUploadArgs[openScriptPath]?.args : undefined} - onRunFormArgsChange={stageRunFormArgs} + readyDataUploadPaths={dataUploads.readyPaths} + runFormInitialArgs={openScriptPath ? dataUploads.argsFor(openScriptPath) : undefined} + onRunFormArgsChange={dataUploads.stage} + inputsOnly={uploadInputsPath !== undefined && uploadInputsPath === openScriptPath} + onRunInputs={runUploadInputs} + onShowScript={() => (uploadInputsPath = undefined)} onSelect={handleCanvasSelect} onAddScriptForAsset={mode === 'edit' ? handleAddScriptForAsset : undefined} onAddPipelineScript={mode === 'edit' ? handleAddPipelineScript : undefined}