feat(pipeline): output-kind picker and per-(lang, output) templates

Add a third stage to PipelineInsertMenu that asks what kind of asset the new script will produce (datatable / ducklake / s3 parquet / s3 object / none). The picked kind drives a real wmill SDK skeleton — typed datatable inserts, ducklake CREATE+INSERT, s3 parquet COPY, etc. — with the upstream asset auto-wired as the input source when added from an asset node. Reorder languages to bun → duckdb → python → sql so data-shaped languages surface first. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-08-23 08:00:45 +00:00 · 2026-05-01 17:26:52 +00:00
parent 621375bc40
commit 79a4cdd884
7 changed files with 720 additions and 264 deletions
@@ -14,6 +14,8 @@
 	} from 'lucide-svelte'
 	import type { ScriptLang } from '$lib/gen'
 	import type { NativeTriggerKind } from './types'
+	import { PIPELINE_LANGUAGES } from './pipelineLanguages'
+	import type { PipelineOutputKind } from './pipelineTemplates'

 	// Each left-column kind is just "pipeline script triggered by <trigger
 	// source>". id === the SCRIPT_TRIGGER_KIND value, so the handler can
@@ -28,7 +30,8 @@
 				path: string,
 				source:
 					| { kind: 'schedule'; cron: string }
-					| { kind: NativeTriggerKind; path: string | undefined }
+					| { kind: NativeTriggerKind; path: string | undefined },
+				outputKind: PipelineOutputKind
 			) => void
 			pathPrefix: string
 			defaultPathSuffix: string
@@ -37,37 +40,28 @@
 	}
 	let { data }: Props = $props()

-	const LANGUAGES: Array<{ label: string; lang: ScriptLang }> = [
-		{ label: 'TypeScript (Bun)', lang: 'bun' },
-		{ label: 'TypeScript (Deno)', lang: 'deno' },
-		{ label: 'Python', lang: 'python3' },
-		{ label: 'PostgreSQL', lang: 'postgresql' },
-		{ label: 'DuckDB', lang: 'duckdb' },
-		{ label: 'BigQuery', lang: 'bigquery' },
-		{ label: 'Snowflake', lang: 'snowflake' },
-		{ label: 'MySQL', lang: 'mysql' },
-		{ label: 'MS SQL', lang: 'mssql' },
-		{ label: 'Bash', lang: 'bash' },
-		{ label: 'Go', lang: 'go' }
-	]
-
 	function handlePick(pick: PipelineInsertPick) {
 		if (!pick.language || !pick.path) return
 		const kindId = pick.kindId as KindId
+		const outputKind = (pick.outputKind ?? 'none') as PipelineOutputKind
 		if (kindId === 'schedule') {
-			data.onAddPipelineScript(pick.language as ScriptLang, pick.path, {
-				kind: 'schedule',
-				cron: data.defaultScheduleCron
-			})
+			data.onAddPipelineScript(
+				pick.language as ScriptLang,
+				pick.path,
+				{ kind: 'schedule', cron: data.defaultScheduleCron },
+				outputKind
+			)
 		} else {
 			// Native trigger reference: user is expected to fill in the
 			// trigger path themselves in the editor (or configure it in the
 			// trigger's own UI). We seed the annotation with an empty ref
 			// the user replaces.
-			data.onAddPipelineScript(pick.language as ScriptLang, pick.path, {
-				kind: kindId,
-				path: undefined
-			})
+			data.onAddPipelineScript(
+				pick.language as ScriptLang,
+				pick.path,
+				{ kind: kindId, path: undefined },
+				outputKind
+			)
 		}
 	}
 </script>
@@ -138,7 +132,8 @@
 			pickLanguage: true
 		}
 	]}
-	languages={LANGUAGES as any}
+	languages={PIPELINE_LANGUAGES as any}
+	pickOutputKind
 	pathPrefix={data.pathPrefix}
 	defaultPathSuffix={data.defaultPathSuffix}
 	onPick={handlePick}
@@ -35,7 +35,8 @@
 		onAddScriptForAsset?: (
 			asset: { kind: AssetKind; path: string },
 			language: import('$lib/gen').ScriptLang,
-			scriptPath: string
+			scriptPath: string,
+			outputKind: import('./pipelineTemplates').PipelineOutputKind
 		) => void
 		// Pipeline-wide + node shown at the top of the graph. Picking any
 		// kind from the menu invokes this one callback with the chosen
@@ -48,7 +49,8 @@
 				| {
 						kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
 						path: string | undefined
-				  }
+				  },
+			outputKind: import('./pipelineTemplates').PipelineOutputKind
 		) => void
 		// Folder-scoped prefix shown as a read-only chip in the insert menu
 		// path input (e.g. `f/{folder}/`). Shared across top + and per-asset +.
@@ -9,6 +9,8 @@
 	import type { ScriptLang } from '$lib/gen'
 	import { workspaceStore } from '$lib/stores'
 	import { sendUserToast } from '$lib/utils'
+	import { PIPELINE_LANGUAGES } from './pipelineLanguages'
+	import type { PipelineOutputKind } from './pipelineTemplates'

 	// Shape used for both the data prop and the run callback. Drafts carry
 	// `content` / `language` so the page-level run handler can dispatch to
@@ -27,7 +29,8 @@
 			onAddScript?: (
 				asset: { kind: AssetKind; path: string },
 				language: ScriptLang,
-				scriptPath: string
+				scriptPath: string,
+				outputKind: PipelineOutputKind
 			) => void
 			pathPrefix?: string
 			defaultPathSuffix?: string
@@ -84,26 +87,13 @@
 		}
 	}

-	const LANGUAGES: Array<{ label: string; lang: ScriptLang }> = [
-		{ label: 'TypeScript (Bun)', lang: 'bun' },
-		{ label: 'TypeScript (Deno)', lang: 'deno' },
-		{ label: 'Python', lang: 'python3' },
-		{ label: 'PostgreSQL', lang: 'postgresql' },
-		{ label: 'DuckDB', lang: 'duckdb' },
-		{ label: 'BigQuery', lang: 'bigquery' },
-		{ label: 'Snowflake', lang: 'snowflake' },
-		{ label: 'MySQL', lang: 'mysql' },
-		{ label: 'MS SQL', lang: 'mssql' },
-		{ label: 'Bash', lang: 'bash' },
-		{ label: 'Go', lang: 'go' }
-	]
-
 	function handlePick(pick: PipelineInsertPick) {
 		if (pick.kindId === 'pipeline_script' && pick.language && pick.path) {
 			data.onAddScript?.(
 				{ kind: data.asset_kind, path: data.path },
 				pick.language as ScriptLang,
-				pick.path
+				pick.path,
+				(pick.outputKind ?? 'none') as PipelineOutputKind
 			)
 		}
 	}
@@ -182,7 +172,8 @@
 						pickLanguage: true
 					}
 				]}
-				languages={LANGUAGES as any}
+				languages={PIPELINE_LANGUAGES as any}
+				pickOutputKind
 				pathPrefix={data.pathPrefix ?? ''}
 				defaultPathSuffix={data.defaultPathSuffix ?? ''}
 				onPick={handlePick}
@@ -18,6 +18,9 @@
 		kindId: string
 		language?: SupportedLanguage
 		path?: string
+		// Picked output asset kind. Optional because some menu instances
+		// (those without `pickOutputKind`) skip that stage entirely.
+		outputKind?: string
 	}
 </script>

@@ -27,6 +30,12 @@
 	import LanguageIcon from '$lib/components/common/languageIcons/LanguageIcon.svelte'
 	import { ArrowLeft, ChevronRight } from 'lucide-svelte'
 	import { tick } from 'svelte'
+	import {
+		PIPELINE_OUTPUT_KINDS,
+		compatibleOutputKinds,
+		type PipelineOutputKind
+	} from './pipelineTemplates'
+	import type { ScriptLang } from '$lib/gen'

 	interface Props {
 		kinds: PipelineInsertKind[]
@@ -38,6 +47,11 @@
 		// Default suffix seeded into the editable input when the user
 		// reaches the path stage (e.g. `new_pipeline_script`).
 		defaultPathSuffix?: string
+		// When true, after the user picks a language we add an output-kind
+		// stage between language and path. The picked kind is forwarded in
+		// onPick(pick.outputKind). When false (or omitted), the menu jumps
+		// directly from language → path, matching the legacy two-stage flow.
+		pickOutputKind?: boolean
 		onPick: (pick: PipelineInsertPick) => void
 		trigger: import('svelte').Snippet
 		placement?: 'bottom' | 'top' | 'left' | 'right'
@@ -48,25 +62,35 @@
 		languages = [],
 		pathPrefix = '',
 		defaultPathSuffix = '',
+		pickOutputKind = false,
 		onPick,
 		trigger: triggerSnippet,
 		placement = 'bottom'
 	}: Props = $props()

-	// Flow stages: kind → lang → path → confirm. `stage` drives the right
-	// column. Only the `language` kinds reach the lang/path stages.
-	// Default-select the first kind, and if it needs a language jump
-	// straight into the lang stage so the user doesn't have to click the
-	// left column to see the language picker.
+	// Flow stages: kind → lang → (output) → path → confirm. `stage` drives
+	// the right column. Only the `language` kinds reach the lang/path
+	// stages. The `output` stage is gated behind `pickOutputKind` so menus
+	// that don't need it (legacy two-column callers) keep their old flow.
 	let selectedKindId = $state<string>(kinds[0]?.id ?? '')
 	let selectedKind = $derived(kinds.find((k) => k.id === selectedKindId) ?? kinds[0])
-	let stage = $state<'lang' | 'path' | 'description'>(
+	let stage = $state<'lang' | 'output' | 'path' | 'description'>(
 		kinds[0]?.pickLanguage ? 'lang' : 'description'
 	)
 	let selectedLanguage = $state<SupportedLanguage | undefined>(undefined)
+	let selectedOutputKind = $state<PipelineOutputKind | undefined>(undefined)
 	let pathSuffix = $state('')
 	let pathInput: HTMLInputElement | undefined = $state(undefined)

+	// Output kinds that have a real template for the picked language. We
+	// hide non-compatible kinds entirely rather than greying them — keeps
+	// the picker scannable and the user never lands on a kind that would
+	// silently fall back to the generic body.
+	let compatibleKinds = $derived.by<PipelineOutputKind[]>(() => {
+		if (!selectedLanguage) return []
+		return compatibleOutputKinds(selectedLanguage as ScriptLang)
+	})
+
 	// Popover closing doesn't unmount its content, so state persists across
 	// opens. Reset everything back to initial state on close so the next
 	// open starts fresh — otherwise reopening lands on the previous path
@@ -76,6 +100,7 @@
 		selectedKindId = kinds[0]?.id ?? ''
 		stage = kinds[0]?.pickLanguage ? 'lang' : 'description'
 		selectedLanguage = undefined
+		selectedOutputKind = undefined
 		pathSuffix = ''
 	}

@@ -88,6 +113,7 @@
 		selectedKindId = k.id
 		stage = 'lang'
 		selectedLanguage = undefined
+		selectedOutputKind = undefined
 		pathSuffix = ''
 	}

@@ -102,24 +128,45 @@
 		return out
 	}

-	async function handleLanguageClick(lang: SupportedLanguage) {
-		selectedLanguage = lang
-		const base = defaultPathSuffix || 'pipeline_script'
-		pathSuffix = `${base}_${shortSlug()}`
-		stage = 'path'
-		// Focus the suffix input so the user can just start typing a name.
+	async function focusPathInput() {
 		await tick()
 		pathInput?.focus()
 		pathInput?.select()
 	}

+	async function handleLanguageClick(lang: SupportedLanguage) {
+		selectedLanguage = lang
+		// If this menu wants an output-kind stage, route through it; the
+		// path suffix only gets seeded once the user has confirmed both
+		// language and output kind.
+		if (pickOutputKind) {
+			selectedOutputKind = undefined
+			stage = 'output'
+			return
+		}
+		const base = defaultPathSuffix || 'pipeline_script'
+		pathSuffix = `${base}_${shortSlug()}`
+		stage = 'path'
+		await focusPathInput()
+	}
+
+	async function handleOutputKindClick(kind: PipelineOutputKind) {
+		selectedOutputKind = kind
+		const base = defaultPathSuffix || 'pipeline_script'
+		pathSuffix = `${base}_${shortSlug()}`
+		stage = 'path'
+		await focusPathInput()
+	}
+
 	function confirmPath(close: () => void) {
 		const suffix = pathSuffix.trim()
 		if (!suffix || !selectedLanguage) return
+		if (pickOutputKind && !selectedOutputKind) return
 		onPick({
 			kindId: selectedKindId,
 			language: selectedLanguage,
-			path: pathPrefix + suffix
+			path: pathPrefix + suffix,
+			outputKind: pickOutputKind ? selectedOutputKind : undefined
 		})
 		close()
 	}
@@ -130,7 +177,7 @@
 			confirmPath(close)
 		} else if (e.key === 'Escape') {
 			e.preventDefault()
-			stage = 'lang'
+			stage = pickOutputKind ? 'output' : 'lang'
 		}
 	}
 </script>
@@ -157,10 +204,19 @@
 	{/snippet}
 	{#snippet content({ close })}
 		{@const singleKind = kinds.length === 1}
+		{@const widthClass = pickOutputKind
+			? singleKind
+				? 'w-[520px]'
+				: 'w-[720px]'
+			: singleKind
+				? 'w-[360px]'
+				: 'w-[560px]'}
 		<div
-			class={singleKind
-				? 'flex flex-row bg-surface-tertiary w-[360px] h-[280px]'
-				: 'flex flex-row divide-x bg-surface-tertiary w-[560px] h-[280px]'}
+			class={[
+				'flex flex-row bg-surface-tertiary h-[280px]',
+				singleKind ? '' : 'divide-x',
+				widthClass
+			].join(' ')}
 		>
 			<!-- Left column: kind picker. Only shown when there's more than one
 			     option — single-kind menus jump straight to language/path. -->
@@ -215,9 +271,9 @@
 							</Button>
 						{/each}
 					</div>
-				{:else if stage === 'path' && selectedLanguage}
-					<div class="flex flex-col gap-3 p-3 grow overflow-auto">
-						<div class="flex items-center gap-2">
+				{:else if stage === 'output' && selectedLanguage}
+					<div class="flex flex-col gap-1 p-2 grow overflow-auto">
+						<div class="flex items-center gap-2 mb-1">
 							<Button
 								variant="subtle"
 								unifiedSize="xs"
@@ -231,6 +287,46 @@
 								<span class="text-xs font-medium">{selectedLanguage}</span>
 							</div>
 						</div>
+						<div class="text-2xs font-normal text-secondary ml-2 mb-1">Output asset</div>
+						{#each PIPELINE_OUTPUT_KINDS.filter((k) => compatibleKinds.includes(k.id)) as k}
+							<button
+								type="button"
+								onclick={() => handleOutputKindClick(k.id)}
+								class="flex flex-col items-start gap-0.5 px-2 py-2 rounded-md text-left transition-colors hover:bg-surface-hover"
+							>
+								<span class="text-sm font-medium leading-tight">{k.label}</span>
+								<span class="text-2xs text-tertiary font-normal leading-snug">
+									{k.description}
+								</span>
+							</button>
+						{/each}
+						{#if compatibleKinds.length === 0}
+							<span class="text-2xs text-tertiary px-2">No output presets for this language.</span>
+						{/if}
+					</div>
+				{:else if stage === 'path' && selectedLanguage}
+					{@const outputMeta = selectedOutputKind
+						? PIPELINE_OUTPUT_KINDS.find((k) => k.id === selectedOutputKind)
+						: undefined}
+					<div class="flex flex-col gap-3 p-3 grow overflow-auto">
+						<div class="flex items-center gap-2">
+							<Button
+								variant="subtle"
+								unifiedSize="xs"
+								startIcon={{ icon: ArrowLeft }}
+								iconOnly
+								title={pickOutputKind ? 'Back to output' : 'Back to language'}
+								onClick={() => (stage = pickOutputKind ? 'output' : 'lang')}
+							/>
+							<div class="flex items-center gap-1.5">
+								<LanguageIcon lang={selectedLanguage} width={13} height={13} />
+								<span class="text-xs font-medium">{selectedLanguage}</span>
+							</div>
+							{#if outputMeta}
+								<span class="text-2xs text-tertiary">·</span>
+								<span class="text-2xs text-secondary">{outputMeta.label}</span>
+							{/if}
+						</div>
 						<div class="flex flex-col gap-1">
 							<span class="text-2xs font-normal text-secondary">Path</span>
 							<!-- Prefix is a separate non-editable <span> sitting next to
@@ -0,0 +1,20 @@
+import type { ScriptLang } from '$lib/gen'
+
+// Pipelines are dataset-shaped, so the menu surfaces the languages users
+// actually reach for first: bun for ergonomic data wrangling, duckdb for
+// in-place SQL on parquet/s3, python for ML/pandas, then the sql dialects
+// for warehouse-resident transforms. Everything else (deno/bash/go) sits
+// below — still creatable, just not the default suggestion.
+export const PIPELINE_LANGUAGES: Array<{ label: string; lang: ScriptLang }> = [
+	{ label: 'TypeScript (Bun)', lang: 'bun' },
+	{ label: 'DuckDB', lang: 'duckdb' },
+	{ label: 'Python', lang: 'python3' },
+	{ label: 'PostgreSQL', lang: 'postgresql' },
+	{ label: 'BigQuery', lang: 'bigquery' },
+	{ label: 'Snowflake', lang: 'snowflake' },
+	{ label: 'MySQL', lang: 'mysql' },
+	{ label: 'MS SQL', lang: 'mssql' },
+	{ label: 'TypeScript (Deno)', lang: 'deno' },
+	{ label: 'Bash', lang: 'bash' },
+	{ label: 'Go', lang: 'go' }
+]
@@ -0,0 +1,502 @@
+import type { ScriptLang, AssetKind } from '$lib/gen'
+
+// What kind of asset the new script will produce. Drives the auto-generated
+// output annotation, the random output path scheme, and the body skeleton.
+//
+// `none` is the conservative default — no output asset annotation, body just
+// has a "fill in" comment. The other kinds inject their respective wmill SDK
+// calls / SQL setup so the script is runnable (modulo schema definition) the
+// moment it's created.
+export type PipelineOutputKind = 'none' | 'datatable' | 'ducklake' | 's3_parquet' | 's3_object'
+
+export type PipelineOutputKindMeta = {
+	id: PipelineOutputKind
+	label: string
+	description: string
+}
+
+// The order here is the order shown in the picker. Datatable + ducklake are
+// the dataset-shaped kinds we want users to gravitate toward; s3 parquet/
+// object are the escape hatches for arbitrary blobs; none is last because
+// picking it disables the whole "auto-generated output" feature.
+export const PIPELINE_OUTPUT_KINDS: PipelineOutputKindMeta[] = [
+	{
+		id: 'datatable',
+		label: 'DataTable',
+		description: 'Postgres-backed typed table'
+	},
+	{
+		id: 'ducklake',
+		label: 'DuckLake',
+		description: 'DuckDB lakehouse table'
+	},
+	{
+		id: 's3_parquet',
+		label: 'S3 Parquet',
+		description: 'Columnar file in object storage'
+	},
+	{
+		id: 's3_object',
+		label: 'S3 Object',
+		description: 'Generic file (JSON/CSV/binary)'
+	},
+	{
+		id: 'none',
+		label: 'No output',
+		description: 'Side-effect only / fill in manually'
+	}
+]
+
+// Per-language compatibility — output kinds that have a hand-rolled template
+// for `lang`. Falling outside this list means the picker grays out the kind
+// (and the generator falls back to `none`-like behavior). The DuckDB lang
+// ducks both datatable + ducklake natively via attached catalogs; the
+// non-postgres warehouse SQL dialects don't get datatable templates because
+// their wmill SDK story doesn't include cross-dialect bridge scripts.
+const LANG_COMPATIBILITY: Record<ScriptLang, PipelineOutputKind[]> = {
+	bun: ['datatable', 'ducklake', 's3_parquet', 's3_object', 'none'],
+	deno: ['datatable', 'ducklake', 's3_parquet', 's3_object', 'none'],
+	python3: ['datatable', 'ducklake', 's3_parquet', 's3_object', 'none'],
+	duckdb: ['datatable', 'ducklake', 's3_parquet', 'none'],
+	postgresql: ['datatable', 'none'],
+	mysql: ['none'],
+	mssql: ['none'],
+	bigquery: ['none'],
+	snowflake: ['none'],
+	oracledb: ['none'],
+	bash: ['s3_object', 'none'],
+	go: ['none'],
+	rust: ['none'],
+	php: ['none'],
+	powershell: ['s3_object', 'none'],
+	nu: ['none'],
+	ansible: ['none'],
+	java: ['none'],
+	csharp: ['none'],
+	graphql: ['none'],
+	bunnative: ['none'],
+	nativets: ['none']
+} as any
+
+export function compatibleOutputKinds(lang: ScriptLang): PipelineOutputKind[] {
+	return LANG_COMPATIBILITY[lang] ?? ['none']
+}
+
+// Generates the random suffix portion of an auto-named output asset. Same
+// 7-char Crockford-ish alphabet we use for script paths so the slugs read
+// consistently.
+function randomSlug(len = 7): string {
+	const a = 'abcdefghijklmnopqrstuvwxyz0123456789'
+	let out = ''
+	for (let i = 0; i < len; i++) out += a[Math.floor(Math.random() * a.length)]
+	return out
+}
+
+// Fully-qualified asset URI for a freshly-created output, given the picked
+// kind. The path is auto-generated to land under `pipelines/<folder>/` for
+// s3 / parquet, and uses the more conventional `main/<name>` form for
+// datatable / ducklake (matches what the schema/preview UI expects). The
+// name suffix is randomized to avoid collisions when the user adds multiple
+// downstream scripts in one session.
+export function autoOutputAsset(
+	kind: PipelineOutputKind,
+	folder: string
+): { kind: AssetKind; path: string } | undefined {
+	const slug = randomSlug()
+	switch (kind) {
+		case 'datatable':
+			return { kind: 'datatable', path: `main/${folder}_${slug}` }
+		case 'ducklake':
+			return { kind: 'ducklake', path: `main/${folder}_${slug}` }
+		case 's3_parquet':
+			return { kind: 's3object', path: `pipelines/${folder}/out_${slug}.parquet` }
+		case 's3_object':
+			return { kind: 's3object', path: `pipelines/${folder}/out_${slug}.json` }
+		case 'none':
+			return undefined
+	}
+}
+
+// URI prefix used in `// on <ref>` annotations — matches ASSET_PREFIX in
+// the page-level handler. Duplicated here so the templates module is
+// self-contained (no circular import with the pipeline page).
+const ASSET_URI_PREFIX: Record<AssetKind, string> = {
+	s3object: 's3://',
+	resource: '$res:',
+	ducklake: 'ducklake://',
+	datatable: 'datatable://',
+	volume: 'volume://'
+}
+
+export function assetUri(asset: { kind: AssetKind; path: string }): string {
+	return `${ASSET_URI_PREFIX[asset.kind]}${asset.path}`
+}
+
+// Comment prefix per language; mirrors what the parser accepts.
+function commentPrefix(lang: ScriptLang): string {
+	switch (lang) {
+		case 'python3':
+		case 'bash':
+		case 'powershell':
+		case 'nu':
+		case 'ansible':
+			return '#'
+		case 'postgresql':
+		case 'mysql':
+		case 'bigquery':
+		case 'snowflake':
+		case 'mssql':
+		case 'oracledb':
+		case 'duckdb':
+			return '--'
+		default:
+			return '//'
+	}
+}
+
+export type DraftTriggerSource =
+	| { kind: 'schedule'; cron: string }
+	| { kind: 'asset'; ref: string }
+	| {
+			kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
+			path: string | undefined
+	  }
+
+export type TemplateContext = {
+	language: ScriptLang
+	outputKind: PipelineOutputKind
+	output?: { kind: AssetKind; path: string }
+	// The upstream asset that triggered creation of this script (if any).
+	// When present, the body loads from it; the `// on <ref>` annotation is
+	// emitted regardless via the trigger source list passed to header().
+	input?: { kind: AssetKind; path: string }
+	triggers: DraftTriggerSource[]
+}
+
+// Header: `// pipeline` + every trigger source as its own annotation line.
+// Output asset is NOT declared here — it's reconstructed from the body's
+// SDK calls / SQL by the asset parser, same as production scripts.
+function header(language: ScriptLang, triggers: DraftTriggerSource[]): string {
+	const p = commentPrefix(language)
+	const lines = triggers.map((t) => {
+		switch (t.kind) {
+			case 'schedule':
+				return `${p} schedule "${t.cron}"`
+			case 'asset':
+				return `${p} on ${t.ref}`
+			default:
+				return `${p} on ${t.kind} ${t.path ?? '<trigger-path>'}`
+		}
+	})
+	return [`${p} pipeline`, ...lines, ''].join('\n')
+}
+
+// Bun / Deno bodies. These share the wmill SDK surface, so we treat them
+// uniformly. The differences vs the previous generic body:
+//   - input asset → real `wmill.loadS3File` / `wmill.datatable(...).fetch()`
+//   - output asset → real `wmill.writeS3File` / `wmill.datatable(...).fetch()`
+//   - no module-scope OUT constant — the parser reads write paths from the
+//     SDK call sites directly, so the constant was redundant.
+function bodyTs(ctx: TemplateContext): string {
+	const { input, output, outputKind } = ctx
+	const importLine = `import * as wmill from "windmill-client"\n`
+
+	const inputBlock = (() => {
+		if (!input) return ''
+		switch (input.kind) {
+			case 's3object':
+				return [
+					`  // Upstream: ${assetUri(input)}`,
+					`  const buf = await wmill.loadS3File({ s3: ${JSON.stringify(input.path)} })`,
+					`  const rows = JSON.parse(new TextDecoder().decode(buf))`,
+					``
+				].join('\n')
+			case 'datatable':
+				return [
+					`  // Upstream: ${assetUri(input)}`,
+					`  const src = wmill.datatable(${JSON.stringify(input.path.split('/')[0] ?? 'main')})`,
+					`  const rows = await src\`SELECT * FROM ${input.path.split('/').slice(1).join('_') || 'table_name'}\`.fetch()`,
+					``
+				].join('\n')
+			case 'ducklake':
+				return [
+					`  // Upstream: ${assetUri(input)}`,
+					`  const lake = wmill.ducklake(${JSON.stringify(input.path.split('/')[0] ?? 'main')})`,
+					`  const rows = await lake\`SELECT * FROM ${input.path.split('/').slice(1).join('_') || 'table_name'}\`.fetch()`,
+					``
+				].join('\n')
+			default:
+				return `  // Upstream: ${assetUri(input)}\n`
+		}
+	})()
+
+	const outputBlock = (() => {
+		if (!output) return `  // (no output asset — fill in side-effect logic)`
+		switch (outputKind) {
+			case 's3_parquet':
+			case 's3_object':
+				return [
+					`  // Output: ${assetUri(output)}`,
+					`  const payload = new TextEncoder().encode(JSON.stringify(rows))`,
+					`  await wmill.writeS3File({ s3: ${JSON.stringify(output.path)} }, payload)`
+				].join('\n')
+			case 'datatable': {
+				const dbName = output.path.split('/')[0] ?? 'main'
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				return [
+					`  // Output: ${assetUri(output)}`,
+					`  const dst = wmill.datatable(${JSON.stringify(dbName)})`,
+					`  await dst\`CREATE TABLE IF NOT EXISTS ${tableName} (id serial primary key, payload jsonb)\`.fetch()`,
+					`  for (const r of rows) {`,
+					`    await dst\`INSERT INTO ${tableName} (payload) VALUES (\${JSON.stringify(r)}::jsonb)\`.fetch()`,
+					`  }`
+				].join('\n')
+			}
+			case 'ducklake': {
+				const dbName = output.path.split('/')[0] ?? 'main'
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				return [
+					`  // Output: ${assetUri(output)}`,
+					`  const lakeOut = wmill.ducklake(${JSON.stringify(dbName)})`,
+					`  await lakeOut\`CREATE TABLE IF NOT EXISTS ${tableName} (payload JSON)\`.fetch()`,
+					`  for (const r of rows) {`,
+					`    await lakeOut\`INSERT INTO ${tableName} VALUES (\${JSON.stringify(r)})\`.fetch()`,
+					`  }`
+				].join('\n')
+			}
+			default:
+				return `  // (fill in)`
+		}
+	})()
+
+	const rowsFallback = !input ? `  const rows: any[] = []\n` : ''
+
+	return [
+		importLine,
+		'export async function main() {',
+		inputBlock,
+		rowsFallback,
+		outputBlock,
+		'}',
+		''
+	].join('\n')
+}
+
+function bodyPython(ctx: TemplateContext): string {
+	const { input, output, outputKind } = ctx
+	const importLine = `import wmill\n`
+
+	const inputBlock = (() => {
+		if (!input) return ''
+		switch (input.kind) {
+			case 's3object':
+				return [
+					`    # Upstream: ${assetUri(input)}`,
+					`    buf = wmill.load_s3_file(${JSON.stringify(input.path)})`,
+					`    import json; rows = json.loads(buf.decode("utf-8"))`
+				].join('\n')
+			case 'datatable':
+				return [
+					`    # Upstream: ${assetUri(input)}`,
+					`    src = wmill.datatable(${JSON.stringify(input.path.split('/')[0] ?? 'main')})`,
+					`    rows = src.query("SELECT * FROM ${input.path.split('/').slice(1).join('_') || 'table_name'}").fetch()`
+				].join('\n')
+			case 'ducklake':
+				return [
+					`    # Upstream: ${assetUri(input)}`,
+					`    lake = wmill.ducklake(${JSON.stringify(input.path.split('/')[0] ?? 'main')})`,
+					`    rows = lake.query("SELECT * FROM ${input.path.split('/').slice(1).join('_') || 'table_name'}").fetch()`
+				].join('\n')
+			default:
+				return `    # Upstream: ${assetUri(input)}`
+		}
+	})()
+
+	const outputBlock = (() => {
+		if (!output) return `    # (no output asset — fill in side-effect logic)`
+		switch (outputKind) {
+			case 's3_parquet':
+			case 's3_object':
+				return [
+					`    # Output: ${assetUri(output)}`,
+					`    import json`,
+					`    wmill.write_s3_file(${JSON.stringify(output.path)}, json.dumps(rows).encode("utf-8"))`
+				].join('\n')
+			case 'datatable': {
+				const dbName = output.path.split('/')[0] ?? 'main'
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				return [
+					`    # Output: ${assetUri(output)}`,
+					`    dst = wmill.datatable(${JSON.stringify(dbName)})`,
+					`    dst.query("CREATE TABLE IF NOT EXISTS ${tableName} (id serial primary key, payload jsonb)").execute()`,
+					`    for r in rows:`,
+					`        dst.query("INSERT INTO ${tableName} (payload) VALUES ($1::jsonb)", json.dumps(r)).execute()`
+				].join('\n')
+			}
+			case 'ducklake': {
+				const dbName = output.path.split('/')[0] ?? 'main'
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				return [
+					`    # Output: ${assetUri(output)}`,
+					`    lake_out = wmill.ducklake(${JSON.stringify(dbName)})`,
+					`    lake_out.query("CREATE TABLE IF NOT EXISTS ${tableName} (payload JSON)").execute()`,
+					`    for r in rows:`,
+					`        lake_out.query("INSERT INTO ${tableName} VALUES ($payload)", payload=json.dumps(r)).execute()`
+				].join('\n')
+			}
+			default:
+				return `    # (fill in)`
+		}
+	})()
+
+	const rowsFallback = !input ? `    rows: list = []` : ''
+
+	return [importLine, 'def main():', inputBlock, rowsFallback, outputBlock, ''].join('\n')
+}
+
+function bodyDuckdb(ctx: TemplateContext): string {
+	const { input, output, outputKind } = ctx
+	const lines: string[] = []
+	if (input) lines.push(`-- Upstream: ${assetUri(input)}`)
+	if (output) lines.push(`-- Output: ${assetUri(output)}`)
+	lines.push('')
+
+	const inSql = (() => {
+		if (!input) return null
+		switch (input.kind) {
+			case 's3object':
+				return `read_parquet('s3://${input.path}')`
+			case 'ducklake': {
+				return null // attach + read shown below
+			}
+			default:
+				return null
+		}
+	})()
+
+	if (input?.kind === 'ducklake') {
+		lines.push(
+			`-- DuckLake catalog is auto-attached as 'lake' inside windmill duckdb scripts.`,
+			`-- Reference upstream tables as lake.<table_name>.`
+		)
+	}
+
+	switch (outputKind) {
+		case 's3_parquet':
+			if (output) {
+				const fromExpr = inSql ?? '(SELECT 1 AS placeholder)'
+				lines.push(
+					`COPY (`,
+					`  SELECT *`,
+					`  FROM ${fromExpr}`,
+					`) TO 's3://${output.path}' (FORMAT 'parquet');`
+				)
+			}
+			break
+		case 'ducklake':
+			if (output) {
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				lines.push(
+					`-- Write into the auto-attached DuckLake catalog.`,
+					`CREATE TABLE IF NOT EXISTS lake.${tableName} AS`,
+					`SELECT * FROM ${inSql ?? '(SELECT 1 AS placeholder)'};`
+				)
+			}
+			break
+		case 'datatable':
+			if (output) {
+				const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+				lines.push(
+					`-- DataTable is exposed as the 'pg' attached database in duckdb scripts.`,
+					`CREATE TABLE IF NOT EXISTS pg.${tableName} AS`,
+					`SELECT * FROM ${inSql ?? '(SELECT 1 AS placeholder)'};`
+				)
+			}
+			break
+		case 'none':
+		default:
+			lines.push(`SELECT 1;`)
+	}
+	lines.push('')
+	return lines.join('\n')
+}
+
+function bodyPostgres(ctx: TemplateContext): string {
+	const { input, output, outputKind } = ctx
+	const lines: string[] = []
+	if (input) lines.push(`-- Upstream: ${assetUri(input)}`)
+	if (output) lines.push(`-- Output: ${assetUri(output)}`)
+	lines.push('')
+
+	if (outputKind === 'datatable' && output) {
+		const tableName = output.path.split('/').slice(1).join('_') || 'table_name'
+		lines.push(
+			`-- The pipeline-script harness runs this query against the`,
+			`-- datatable backing the configured workspace storage.`,
+			`CREATE TABLE IF NOT EXISTS ${tableName} (`,
+			`  id serial primary key,`,
+			`  payload jsonb,`,
+			`  computed_at timestamptz default now()`,
+			`);`,
+			``,
+			`-- Replace with your real INSERT.`,
+			`INSERT INTO ${tableName} (payload) VALUES ('{}'::jsonb);`
+		)
+	} else {
+		lines.push(`SELECT 1;`)
+	}
+	lines.push('')
+	return lines.join('\n')
+}
+
+function bodyBash(ctx: TemplateContext): string {
+	const { input, output, outputKind } = ctx
+	const lines: string[] = []
+	if (input) lines.push(`# Upstream: ${assetUri(input)}`)
+	if (output) lines.push(`# Output: ${assetUri(output)}`)
+	if (outputKind === 's3_object' && output) {
+		lines.push(
+			``,
+			`# Use the wmill CLI / curl to upload to ${output.path}`,
+			`echo '{}' > /tmp/out.json`
+		)
+	} else {
+		lines.push(``, `true`)
+	}
+	lines.push('')
+	return lines.join('\n')
+}
+
+function genericBody(ctx: TemplateContext): string {
+	const p = commentPrefix(ctx.language)
+	const lines: string[] = []
+	if (ctx.input) lines.push(`${p} Upstream: ${assetUri(ctx.input)}`)
+	if (ctx.output) lines.push(`${p} Output: ${assetUri(ctx.output)}`)
+	lines.push(`${p} Fill in pipeline logic.`)
+	return lines.join('\n') + '\n'
+}
+
+// Entry point: returns the full source (header + body) for the new draft.
+// The returned content is ready to drop into a Script as `content` — no
+// further mutation needed, including for the trigger annotations.
+export function generatePipelineDraft(ctx: TemplateContext): string {
+	const head = header(ctx.language, ctx.triggers)
+	const body = (() => {
+		switch (ctx.language) {
+			case 'bun':
+			case 'deno':
+				return bodyTs(ctx)
+			case 'python3':
+				return bodyPython(ctx)
+			case 'duckdb':
+				return bodyDuckdb(ctx)
+			case 'postgresql':
+				return bodyPostgres(ctx)
+			case 'bash':
+				return bodyBash(ctx)
+			default:
+				return genericBody(ctx)
+		}
+	})()
+	return head + body
+}
@@ -15,6 +15,12 @@
 		parsePipelineAnnotations,
 		type PipelineAnnotations
 	} from '$lib/components/assets/AssetGraph/parsePipelineAnnotations'
+	import {
+		generatePipelineDraft,
+		autoOutputAsset,
+		type PipelineOutputKind,
+		type DraftTriggerSource
+	} from '$lib/components/assets/AssetGraph/pipelineTemplates'
 	import { decodeState, encodeState } from '$lib/utils'
 	import { onMount, untrack } from 'svelte'
 	import {
@@ -27,7 +33,6 @@
 		RefreshCw
 	} from 'lucide-svelte'
 	import { JobService, OpenAPI, type AssetKind, type Script, type ScriptLang } from '$lib/gen'
-	import { initialCode } from '$lib/script_helpers'
 	import { resource } from 'runed'
 	import { Pane, Splitpanes } from 'svelte-splitpanes'
 	import { emptySchema } from '$lib/utils'
@@ -70,7 +75,10 @@
 	// user can come back to it.
 	type Draft = {
 		script: Script
-		outputAsset: { kind: AssetKind; path: string }
+		// Undefined when the user picked `outputKind === 'none'` — the draft
+		// has no auto-generated output asset, so the graph overlay skips
+		// synthesizing a write edge for it.
+		outputAsset?: { kind: AssetKind; path: string }
 	}
 	let drafts = $state<Map<string, Draft>>(new Map())

@@ -153,190 +161,24 @@
 		}
 	})

-	// Crockford-ish random slug for asset paths. 7 chars of [a-z0-9] gives
-	// ~36^7 ≈ 7.8e10 combinations — collision-free in practice for the
-	// handful of scripts a user creates in a session.
-	function randomSlug(len = 7): string {
-		const alphabet = 'abcdefghijklmnopqrstuvwxyz0123456789'
-		let out = ''
-		for (let i = 0; i < len; i++) {
-			out += alphabet[Math.floor(Math.random() * alphabet.length)]
-		}
-		return out
-	}
-
-	function randomOutputAssetPath(): { kind: AssetKind; path: string } {
-		// `s3://` is the most universal kind; works as a literal across every
-		// supported language without extra imports. Inside `pipelines/<folder>/`
-		// so outputs of a single pipeline cluster under one prefix.
-		return { kind: 's3object', path: `pipelines/${folder}/out_${randomSlug()}.parquet` }
-	}
-
-	// Language → comment prefix recognized by parse_pipeline_annotations.
-	function commentPrefix(lang: ScriptLang): string {
-		switch (lang) {
-			case 'python3':
-			case 'bash':
-			case 'powershell':
-			case 'nu':
-			case 'ansible':
-				return '#'
-			case 'postgresql':
-			case 'mysql':
-			case 'bigquery':
-			case 'snowflake':
-			case 'mssql':
-			case 'oracledb':
-			case 'duckdb':
-				return '--'
-			default:
-				return '//'
-		}
-	}
-
-	// Each entry becomes a `// on <…>` line in the seeded template. Multiple
-	// trigger sources are valid (a script can be fired by both a schedule
-	// and a webhook, for instance), but the menu only seeds one at a time.
-	type DraftTriggerSource =
-		| { kind: 'schedule'; cron: string }
-		| { kind: 'asset'; ref: string } // already-prefixed (e.g. s3://…)
-		| {
-				kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
-				path: string | undefined
-		  }
-
-	// Annotation block prepended to every new draft. Pipeline scripts opt in
-	// with `// pipeline` (strict — keyword on its own line) plus zero or
-	// more trigger declarations. We deliberately don't include a doc block
-	// here: it's noise on the editor canvas and the annotation grammar is
-	// documented in /docs.
-	function pipelineHeader(language: ScriptLang, sources: DraftTriggerSource[]): string {
-		const p = commentPrefix(language)
-		const triggerLines = sources.map((s) => {
-			switch (s.kind) {
-				case 'schedule':
-					// Schedule is a top-level annotation, not under `on`.
-					return `${p} schedule "${s.cron}"`
-				case 'asset':
-					return `${p} on ${s.ref}`
-				default:
-					// Empty placeholder path makes it visible the user needs
-					// to fill in the trigger reference.
-					return `${p} on ${s.kind} ${s.path ?? '<trigger-path>'}`
-			}
-		})
-		return [`${p} pipeline`, ...triggerLines, ''].join('\n')
-	}
-
-	// Minimal pipeline-script body per language. No user-facing args (pipeline
-	// scripts are fired by triggers, not invoked interactively); the runtime
-	// auto-fills partition/context fields when applicable. The output URI is
-	// declared as a module-scope literal so the backend asset parser sees the
-	// write on deploy and the graph stays in sync.
-	function pipelineBody(language: ScriptLang, outputAssetUri?: string): string {
-		const out = outputAssetUri
-		switch (language) {
-			case 'python3':
-				return [
-					out ? `OUT = "${out}"\n` : '',
-					'def main():',
-					'    # Read inputs and write to OUT.',
-					'    pass',
-					''
-				].join('\n')
-			case 'bun':
-			case 'deno':
-				return [
-					out ? `const OUT = "${out}"\n` : '',
-					'export async function main() {',
-					'    // Read inputs and write to OUT.',
-					'}',
-					''
-				].join('\n')
-			case 'postgresql':
-			case 'mysql':
-			case 'bigquery':
-			case 'snowflake':
-			case 'mssql':
-			case 'oracledb':
-				return [
-					out ? `-- writes to: ${out}` : '',
-					"-- Read with FROM 's3://...', write with INSERT / COPY.",
-					'SELECT 1;',
-					''
-				]
-					.filter(Boolean)
-					.join('\n')
-			case 'duckdb':
-				return [
-					out ? `-- writes to: ${out}` : '',
-					'-- Read with read_parquet/read_csv on s3:// paths;',
-					"-- write with COPY (...) TO 's3://...'.",
-					'SELECT 1;',
-					''
-				]
-					.filter(Boolean)
-					.join('\n')
-			case 'bash':
-				return [out ? `OUT="${out}"\n` : '', '# Read inputs and write to "$OUT".', 'true', ''].join(
-					'\n'
-				)
-			case 'powershell':
-				return [out ? `$OUT = "${out}"\n` : '', '# Read inputs and write to $OUT.', ''].join('\n')
-			case 'nu':
-				return [out ? `let OUT = "${out}"\n` : '', '# Read inputs and write to $OUT.', ''].join(
-					'\n'
-				)
-			case 'go':
-				return [
-					'package inner',
-					'',
-					out ? `const OUT = "${out}"\n` : '',
-					'func main() (interface{}, error) {',
-					'    // Read inputs and write to OUT.',
-					'    return nil, nil',
-					'}',
-					''
-				].join('\n')
-			case 'rust':
-				return [
-					out ? `const OUT: &str = "${out}";\n` : '',
-					'fn main() -> anyhow::Result<()> {',
-					'    // Read inputs and write to OUT.',
-					'    Ok(())',
-					'}',
-					''
-				].join('\n')
-			case 'ansible':
-				return [
-					'---',
-					out ? `# writes to: ${out}` : '',
-					'- name: Pipeline play',
-					'  hosts: localhost',
-					'  tasks: []',
-					''
-				]
-					.filter(Boolean)
-					.join('\n')
-			default:
-				// Conservative fallback for languages without a hand-rolled
-				// minimal template — windmill ships a default body that we
-				// can use rather than ship something broken. Only loses the
-				// argless guarantee for these languages; users can trim args
-				// themselves.
-				return initialCode(language as any, 'script', 'script')
-		}
-	}
-
+	// Build a runnable Script from picked language / triggers / output.
+	// Delegates to the shared template generator (pipelineTemplates.ts) so
+	// the same logic is reachable from anywhere a draft is needed.
 	function buildDraft(
 		language: ScriptLang,
 		scriptPath: string,
-		sources: DraftTriggerSource[],
-		outputAssetUri?: string
+		triggers: DraftTriggerSource[],
+		outputKind: PipelineOutputKind,
+		output: { kind: AssetKind; path: string } | undefined,
+		input: { kind: AssetKind; path: string } | undefined
 	): Script {
-		const header = pipelineHeader(language, sources)
-		const body = pipelineBody(language, outputAssetUri)
-		const content = header + body
+		const content = generatePipelineDraft({
+			language,
+			outputKind,
+			output,
+			input,
+			triggers
+		})
 		// Cast through unknown: the Script generated type has many readonly
 		// deployment fields (hash, created_*) that we don't care about for a
 		// local draft. The details pane only reads path/language/content/schema.
@@ -362,13 +204,16 @@
 	function openMaterializerDraft(
 		language: ScriptLang,
 		scriptPath: string,
-		sources: DraftTriggerSource[]
+		triggers: DraftTriggerSource[],
+		outputKind: PipelineOutputKind,
+		input?: { kind: AssetKind; path: string }
 	) {
-		const out = randomOutputAssetPath()
-		const outputUri = `${ASSET_PREFIX[out.kind]}${out.path}`
-		const script = buildDraft(language, scriptPath, sources, outputUri)
+		const out = autoOutputAsset(outputKind, folder)
+		const script = buildDraft(language, scriptPath, triggers, outputKind, out, input)
 		// Write the new draft into the map (structural update so Svelte
-		// re-derives graphWithDraft) and focus it in the details pane.
+		// re-derives graphWithDraft) and focus it in the details pane. When
+		// the user picked `none`, `outputAsset` is undefined and the graph
+		// overlay skips synthesizing a write edge.
 		const next = new Map(drafts)
 		next.set(scriptPath, { script, outputAsset: out })
 		drafts = next
@@ -417,16 +262,18 @@
 				unsaved: true
 			})
 			const out = d.outputAsset
-			const hasAsset = assets.some((a) => a.kind === out.kind && a.path === out.path)
-			if (!hasAsset) assets.push({ kind: out.kind, path: out.path })
-			edges.push({
-				runnable_path: path,
-				runnable_kind: 'script',
-				asset_kind: out.kind,
-				asset_path: out.path,
-				access_type: 'w',
-				unsaved: true
-			})
+			if (out) {
+				const hasAsset = assets.some((a) => a.kind === out.kind && a.path === out.path)
+				if (!hasAsset) assets.push({ kind: out.kind, path: out.path })
+				edges.push({
+					runnable_path: path,
+					runnable_kind: 'script',
+					asset_kind: out.kind,
+					asset_path: out.path,
+					access_type: 'w',
+					unsaved: true
+				})
+			}
 			// Seed trigger edges (schedule + asset) from the draft's template
 			// so the graph stays stable when the user clicks off this draft.
 			// Live annotations (below) take over for the currently-open draft
@@ -781,12 +628,15 @@
 									selection = s
 								}
 							}}
-							onAddScriptForAsset={(asset, language, scriptPath) => {
+							onAddScriptForAsset={(asset, language, scriptPath, outputKind) => {
 								const ref = `${ASSET_PREFIX[asset.kind]}${asset.path}`
-								openMaterializerDraft(language, scriptPath, [{ kind: 'asset', ref }])
+								openMaterializerDraft(language, scriptPath, [{ kind: 'asset', ref }], outputKind, {
+									kind: asset.kind,
+									path: asset.path
+								})
 							}}
-							onAddPipelineScript={(language, scriptPath, source) =>
-								openMaterializerDraft(language, scriptPath, [source])}
+							onAddPipelineScript={(language, scriptPath, source, outputKind) =>
+								openMaterializerDraft(language, scriptPath, [source], outputKind)}
 							onRunProducer={async (producer) => {
 								// Saved scripts go through runScriptByPath; drafts
 								// have no DB row yet, so dispatch to runScriptPreview