feat: AI attaches trigger drafts to the pipeline nodes that declare them; materialize nodes keep no stale output

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts

@@ -20,7 +20,9 @@ function makeHandle(
 		getDrafts: () => drafts,
 		setDrafts: (next) => (drafts = next),
 		newDraftLocalId: () => 'id',
-		onForgetPath: (p) => forgotten.push(p)
+		onForgetPath: (p) => forgotten.push(p),
+		getTriggerDrafts: () => new Map(),
+		setTriggerDraft: () => true
 	})
 	return { handle, drafts: () => drafts, forgotten: () => forgotten }
 }
@@ -155,6 +157,30 @@ describe('pipeline AI direct-draft helpers', () => {
 		expect(res.detectedWrites).toEqual(['ducklake://main/out'])
 	})
 
+	// The canvas draws an inactive draft's stored outputs beside its materialize
+	// target, so a seeded default left on a materialize node shows a phantom asset.
+	it('keeps no seeded or stale output on a node with a materialize target', async () => {
+		vi.spyOn(ScriptService, 'getScriptByPath').mockRejectedValue(new Error('404'))
+		const { handle, drafts } = makeHandle()
+		await handle.proposeNode({
+			path: 'f/x/mat',
+			language: 'duckdb' as any,
+			content: '-- pipeline\n-- materialize ducklake://main/abc\nSELECT 1',
+			outputKind: 'ducklake' as any
+		})
+		expect(drafts().get('f/x/mat')?.outputAssets).toBeUndefined()
+
+		await handle.proposeNode({
+			path: 'f/x/later',
+			language: 'duckdb' as any,
+			content: '-- pipeline\nSELECT 1',
+			outputKind: 'ducklake' as any
+		})
+		expect(drafts().get('f/x/later')?.outputAssets?.length).toBeGreaterThan(0)
+		await handle.editNode('f/x/later', '-- pipeline\n-- materialize ducklake://main/abc\nSELECT 1')
+		expect(drafts().get('f/x/later')?.outputAssets).toBeUndefined()
+	})
+
 	it('editNode rejects a path outside the open folder', async () => {
 		const { handle, drafts } = makeHandle()
 		await expect(handle.editNode('f/other/foo', '-- pipeline')).rejects.toThrow(/open folder/)

frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts

@@ -9,7 +9,12 @@ import {
 import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
 import { assetUri, autoOutputAsset, type PipelineOutputKind } from './pipelineTemplates'
 import { parsePipelineAnnotations, scd2CurrentTargetPath } from './parsePipelineAnnotations'
-import type { AssetGraphResponse } from './types'
+import type {
+	AssetGraphResponse,
+	PipelineTriggerDraft,
+	PipelineTriggerDraftKind
+} from './types'
+import { defaultTriggerPath, isDraftableTriggerKind, triggerDraftKey } from './pipelineTriggerDrafts'
 import type {
 	PipelineAIChatHelpers,
 	PipelineContext,
@@ -64,6 +69,9 @@ export type PipelineAiHelperDeps = {
 	onRunStarted?: (jobId: string, path: string) => void
 	/** Whether the folder has trigger drafts, which live beside the script drafts. */
 	hasTriggerDrafts?: () => boolean
+	getTriggerDrafts: () => Map<string, PipelineTriggerDraft>
+	/** False when another trigger draft already holds the config's path. */
+	setTriggerDraft: (d: PipelineTriggerDraft, previousKey?: string) => boolean
 }
 
 export function makePipelineScript(
@@ -215,6 +223,29 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 		}
 	}
 
+	// Declared `on <kind>` bindings of a node that a trigger row or draft fills.
+	// Webhook and data-upload need none, so they never count as unconfigured.
+	function unconfiguredTriggers(path: string, content: string): PipelineTriggerDraftKind[] {
+		const declared = parsePipelineAnnotations(content)
+			.nativeTriggers.map((n) => n.kind)
+			.filter(isDraftableTriggerKind)
+		const drafted = new Set(
+			[...deps.getTriggerDrafts().values()]
+				.filter((d) => d.config.script_path === path)
+				.map((d) => d.kind)
+		)
+		const deployed = new Set(
+			deps
+				.getResolvedGraph()
+				.triggers.filter(
+					(t) =>
+						t.trigger_kind !== 'asset' && t.runnable_path === path && t.path && !t.missing
+				)
+				.map((t) => t.trigger_kind)
+		)
+		return [...new Set(declared)].filter((k) => !drafted.has(k) && !deployed.has(k))
+	}
+
 	const helpers: PipelineAIChatHelpers = {
 		getFolder: folderName,
 		hasDrafts: () => deps.getDrafts().size > 0 || (deps.hasTriggerDrafts?.() ?? false),
@@ -271,10 +302,13 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 			}
 			const inferred = await inferDraftAssets(language, content)
 			// Fall back to a seeded output (from the declared output_kind) when the
-			// body doesn't yet write anything inferable.
+			// body doesn't yet write anything inferable. A `materialize` target is the
+			// output already (the canvas adds it from the annotation): a seed beside it
+			// would show as a second, default-named output.
+			const materialized = materializeWrites(content).length > 0
 			const seeded =
 				inferred.writes[0] ??
-				(outputKind
+				(outputKind && !materialized
 					? autoOutputAsset(outputKind as PipelineOutputKind, folderName(), language)
 					: undefined)
 			// Effective outputs = what actually becomes an output edge on the canvas:
@@ -323,7 +357,15 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 				baseScript = await ScriptService.getScriptByPath({ workspace, path })
 			}
 			const inferred = await inferDraftAssets(baseScript.language, content)
-			const outputAssets = inferred.writes.length > 0 ? inferred.writes : existing?.outputAssets
+			// The stored outputs only stand in for a body that writes nothing inferable;
+			// with a `materialize` target the annotation is the output, so stale ones
+			// (a seeded default name, the target before a rename) must not linger.
+			const outputAssets =
+				inferred.writes.length > 0
+					? inferred.writes
+					: materializeWrites(content).length > 0
+						? undefined
+						: existing?.outputAssets
 			const next = new Map(drafts)
 			next.set(path, {
 				localId: existing?.localId ?? deps.newDraftLocalId(),
@@ -342,6 +384,53 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 				)
 			}
 		},
+		unconfiguredTriggers: async (path) => {
+			const body = await helpers.getNodeBody(path)
+			return body ? unconfiguredTriggers(path, body.content) : []
+		},
+		setNodeTrigger: async (path, kind, config) => {
+			deps.ensureEditable?.()
+			assertInFolder(path)
+			const body = await helpers.getNodeBody(path)
+			if (!body) throw new Error(`No pipeline node at '${path}'. Build it first.`)
+			const prefix = body.language === 'python3' || body.language === 'bash' ? '#' : body.language === 'duckdb' || body.language === 'postgresql' ? '--' : '//'
+			if (!parsePipelineAnnotations(body.content).nativeTriggers.some((n) => n.kind === kind)) {
+				throw new Error(
+					`'${path}' does not declare \`${prefix} on ${kind}\`. Add that annotation line with edit_pipeline_node first, so the trigger is wired to the node.`
+				)
+			}
+			const row = deps
+				.getResolvedGraph()
+				.triggers.find(
+					(t) =>
+						t.trigger_kind === kind &&
+						t.runnable_path === path &&
+						t.path &&
+						!t.missing &&
+						!(t as { draft?: boolean }).draft
+				)
+			if (row && 'path' in row) {
+				throw new Error(
+					`'${path}' already has a deployed ${kind} trigger at '${row.path}'. Change that one instead of adding a second.`
+				)
+			}
+			const existing = [...deps.getTriggerDrafts().values()].find(
+				(d) => d.kind === kind && d.config.script_path === path
+			)
+			const triggerPath =
+				(typeof config.path === 'string' && config.path) ||
+				existing?.config.path ||
+				defaultTriggerPath(path, kind)
+			const saved = deps.setTriggerDraft(
+				{ kind, config: { ...config, path: triggerPath, script_path: path, is_flow: false } },
+				existing ? triggerDraftKey(kind, existing.config.path) : undefined
+			)
+			if (!saved) {
+				throw new Error(`Another draft trigger already uses the path '${triggerPath}'. Pass a different config.path.`)
+			}
+			deps.onShowDrafts?.()
+			return { path: triggerPath, replaced: !!existing }
+		},
 		removeProposedNode: async (path) => {
 			if (!deps.getDrafts().has(path)) {
 				throw new Error(`No unsaved draft at '${path}' to discard.`)

frontend/src/lib/components/copilot/chat/AIChatManager.test.ts

@@ -4336,6 +4336,8 @@ describe('AIChatManager.waitForPipelineHelpers', () => {
 			proposeNode: async () => ({ path: '', detectedReads: [], detectedWrites: [] }),
 			editNode: async () => ({ detectedReads: [], detectedWrites: [] }),
 			removeProposedNode: async () => {},
+			unconfiguredTriggers: async () => [],
+			setNodeTrigger: async () => ({ path: '', replaced: false }),
 			testNode: async () => undefined
 		}
 	}

frontend/src/lib/components/copilot/chat/pipeline/core.test.ts

@@ -66,6 +66,11 @@ function makeHelpers(overrides: Partial<PipelineAIChatHelpers> = {}): {
 			return { detectedReads: [], detectedWrites: [] }
 		},
 		removeProposedNode: record('removeProposedNode'),
+		unconfiguredTriggers: async () => [],
+		setNodeTrigger: async (path, kind, config) => {
+			calls.setNodeTrigger = [...(calls.setNodeTrigger ?? []), [path, kind, config]]
+			return { path: `${path}_${kind}`, replaced: false }
+		},
 		testNode: async () => 'job-123',
 		...overrides
 	}
@@ -80,6 +85,7 @@ describe('pipeline tools', () => {
 			'get_pipeline_graph',
 			'read_pipeline_node',
 			'remove_pipeline_node',
+			'set_pipeline_trigger',
 			'test_pipeline_node'
 		])
 	})
@@ -124,6 +130,49 @@ describe('pipeline tools', () => {
 		expect(out).toContain('not deployed')
 	})
 
+	it('build_pipeline_node attaches declared triggers, and flags one left unset', async () => {
+		const { helpers, calls } = makeHelpers({ unconfiguredTriggers: async () => ['kafka'] })
+		const out = await toolByName('build_pipeline_node').fn({
+			args: {
+				path: 'f/analytics/ingest',
+				language: 'python3',
+				content: '# pipeline\n# on schedule\n# on kafka\ndef main():\n    return 1',
+				triggers: [{ kind: 'schedule', config: { schedule: '0 0 6 * * *', timezone: 'UTC' } }]
+			},
+			workspace: 'w',
+			helpers,
+			toolCallbacks: noopCallbacks(),
+			toolId: 't'
+		})
+		expect(calls.setNodeTrigger?.[0]).toEqual([
+			'f/analytics/ingest',
+			'schedule',
+			{ args: {}, enabled: true, schedule: '0 0 6 * * *', timezone: 'UTC' }
+		])
+		expect(out).toContain("schedule 'f/analytics/ingest_schedule'")
+		expect(out).toContain('ACTION REQUIRED')
+		expect(out).toContain('set_pipeline_trigger')
+	})
+
+	it('build_pipeline_node rejects an invalid trigger config before staging the node', async () => {
+		const { helpers, calls } = makeHelpers()
+		await expect(
+			toolByName('build_pipeline_node').fn({
+				args: {
+					path: 'f/analytics/ingest',
+					language: 'python3',
+					content: '# pipeline\n# on schedule\ndef main():\n    return 1',
+					triggers: [{ kind: 'schedule', config: { timezone: 'UTC' } }]
+				},
+				workspace: 'w',
+				helpers,
+				toolCallbacks: noopCallbacks(),
+				toolId: 't'
+			})
+		).rejects.toThrow(/schedule/)
+		expect(calls.proposeNode).toBeUndefined()
+	})
+
 	it('build_pipeline_node reports the inferred asset lineage', async () => {
 		const { helpers } = makeHelpers({
 			proposeNode: async (input) => ({

frontend/src/lib/components/copilot/chat/pipeline/core.ts

@@ -11,6 +11,9 @@ import {
 	type SessionTool
 } from '../sessionCapabilities'
 import type { PipelineOutputKind } from '$lib/components/assets/AssetGraph/pipelineTemplates'
+import type { PipelineTriggerDraftKind } from '$lib/components/assets/AssetGraph/types'
+import { DRAFTABLE_TRIGGER_KINDS } from '$lib/components/assets/AssetGraph/pipelineTriggerDrafts'
+import { scheduleRequestSchema, triggerRequestSchemas } from '../workspaceToolsZod.gen'
 import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
 
 // ============================================================================
@@ -83,6 +86,15 @@ export interface PipelineAIChatHelpers {
 	) => Promise<{ detectedReads: string[]; detectedWrites: string[] }>
 	/** Discard the unsaved draft at a path (undo a build_pipeline_node). */
 	removeProposedNode: (path: string) => Promise<void>
+	/** Trigger kinds the node declares (`on <kind>`) that no trigger row or draft fills. */
+	unconfiguredTriggers: (path: string) => Promise<PipelineTriggerDraftKind[]>
+	/** Attach (or replace) the node's trigger draft of that kind, deployed with the
+	 * pipeline. `config` is the kind's trigger config; path/script_path are filled in. */
+	setNodeTrigger: (
+		path: string,
+		kind: PipelineTriggerDraftKind,
+		config: Record<string, any>
+	) => Promise<{ path: string; replaced: boolean }>
 	/** Preview-run a node (draft body preferred). Returns the started job id. */
 	testNode: (path: string, args?: Record<string, any>) => Promise<string | undefined>
 }
@@ -183,6 +195,61 @@ const readPipelineNodeToolDef = createToolDef(
 // Mutation tools (apply directly as unsaved drafts; never deploy)
 // ----------------------------------------------------------------------------
 
+const triggerKindEnum = z.enum(DRAFTABLE_TRIGGER_KINDS as [PipelineTriggerDraftKind, ...PipelineTriggerDraftKind[]])
+
+const nodeTriggerSchema = z.object({
+	kind: triggerKindEnum.describe('Trigger kind; the node must declare `on <kind>`.'),
+	config: z
+		.record(z.string(), z.any())
+		.describe(
+			'The trigger configuration. schedule: { schedule (6-field cron, e.g. "0 0 6 * * *"), timezone (IANA, e.g. "UTC"), args? (the node\'s inputs), summary? }. Any other kind: the fields get_trigger_schema returns for it. path, script_path and is_flow are filled in; pass path only to name the trigger yourself.'
+		)
+})
+
+const setPipelineTriggerSchema = nodeTriggerSchema.extend({
+	path: z.string().describe('Workspace path of the node the trigger runs.')
+})
+
+const setPipelineTriggerToolDef = createToolDef(
+	setPipelineTriggerSchema,
+	'set_pipeline_trigger',
+	'Attach the trigger a node declares with `on <schedule|email|kafka|mqtt|amqp|nats|postgres|sqs|gcp>` (its cron, topic, queue…) as an unsaved trigger draft, deployed together with the pipeline. Calling it again for the same node and kind replaces that draft. Webhook and data_upload need no trigger.',
+	{ strict: false }
+)
+
+/** Validates a node trigger's config against the kind's request schema, with the
+ * fields the pipeline fills in present so they are not reported missing. */
+function validateTriggerConfig(
+	kind: PipelineTriggerDraftKind,
+	config: Record<string, any>,
+	nodePath: string
+): Record<string, any> {
+	const filled = { path: `${nodePath}_${kind}`, script_path: nodePath, is_flow: false, ...config }
+	if (kind === 'schedule') {
+		const parsed = scheduleRequestSchema.safeParse({ args: {}, enabled: true, ...filled })
+		if (!parsed.success) {
+			throw new Error(
+				`Invalid schedule config: ${parsed.error.issues.map((i) => `${i.path.join('.') || '<root>'}: ${i.message}`).join('; ')}`
+			)
+		}
+		return { args: {}, enabled: true, ...config }
+	}
+	const parsed = triggerRequestSchemas[kind].safeParse(filled)
+	if (!parsed.success) {
+		throw new Error(
+			`Invalid config for a "${kind}" trigger. Call get_trigger_schema with kind "${kind}" for its exact fields. Issues: ${parsed.error.issues.map((i) => `${i.path.join('.') || '<root>'}: ${i.message}`).join('; ')}`
+		)
+	}
+	return config
+}
+
+/** The note every node write ends with while a declared trigger is still unset: the
+ * node would deploy with nothing to start it. */
+function unconfiguredTriggersNote(path: string, kinds: readonly string[]): string {
+	if (kinds.length === 0) return ''
+	return ` ACTION REQUIRED: '${path}' declares \`on ${kinds.join('`, `on ')}\` but has no ${kinds.join('/')} trigger yet, so nothing would start it once deployed. Call set_pipeline_trigger for it now (a schedule needs its cron and timezone; ask the user only if the cadence cannot be inferred).`
+}
+
 const buildPipelineNodeSchema = z.object({
 	path: z
 		.string()
@@ -197,7 +264,13 @@ const buildPipelineNodeSchema = z.object({
 		.describe(
 			"Full script source. Start it with the `pipeline` annotation as a top-of-file comment in the LANGUAGE'S comment syntax — `-- pipeline` for SQL (duckdb/postgresql), `# pipeline` for python3/bash, `// pipeline` for bun/TS — to mark it a pipeline member; declare inputs the same way (e.g. `-- on <asset-uri|schedule|webhook|...>`), and write outputs via the wmill SDK / SQL so the lineage edges are inferred. A `// pipeline` line in a SQL node is a syntax error. Read existing node bodies first to match conventions."
 		),
-	output_kind: outputKindSchema.optional()
+	output_kind: outputKindSchema.optional(),
+	triggers: z
+		.array(nodeTriggerSchema)
+		.optional()
+		.describe(
+			'Triggers for the `on <kind>` lines the content declares (schedule, email, kafka, …), attached as trigger drafts in the same step. Every such declaration needs one; webhook and data_upload need none.'
+		)
 })
 
 const buildPipelineNodeToolDef = createToolDef(
@@ -314,8 +387,14 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
 		showDetails: true,
 		showFade: true,
 		fn: async ({ args, helpers, toolId, toolCallbacks }) => {
-			const { path, language, content, output_kind } = buildPipelineNodeSchema.parse(args)
+			const { path, language, content, output_kind, triggers } =
+				buildPipelineNodeSchema.parse(args)
 			const pipeline = pipelineForPath(helpers, path)
+			// Validated before the node is staged, so a bad config leaves nothing half-built.
+			const configs = (triggers ?? []).map((t) => ({
+				kind: t.kind,
+				config: validateTriggerConfig(t.kind, t.config, path)
+			}))
 			toolCallbacks.setToolStatus(toolId, { content: `Building node '${path}'...` })
 			const { detectedReads, detectedWrites } = await pipeline.proposeNode({
 				path,
@@ -323,12 +402,20 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
 				content,
 				outputKind: output_kind as PipelineOutputKind | undefined
 			})
+			const attached: string[] = []
+			for (const t of configs) {
+				const { path: triggerPath } = await pipeline.setNodeTrigger(path, t.kind, t.config)
+				attached.push(`${t.kind} '${triggerPath}'`)
+			}
 			recordPipelineModified(toolCallbacks, pipeline)
 			toolCallbacks.setToolStatus(toolId, {
 				content: `Added draft node '${path}'`,
 				result: 'Success'
 			})
-			return `Pipeline node '${path}' added as an unsaved draft on the canvas. It is not deployed — the user deploys it.${inferredLineageNote(detectedReads, detectedWrites)}`
+			const attachedNote = attached.length
+				? ` Attached trigger draft${attached.length > 1 ? 's' : ''}: ${attached.join(', ')}.`
+				: ''
+			return `Pipeline node '${path}' added as an unsaved draft on the canvas. It is not deployed — the user deploys it.${attachedNote}${inferredLineageNote(detectedReads, detectedWrites)}${unconfiguredTriggersNote(path, await pipeline.unconfiguredTriggers(path))}`
 		}
 	},
 	{
@@ -358,7 +445,25 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
 				content: `Edited draft '${path}'`,
 				result: 'Success'
 			})
-			return `Pipeline node '${path}' updated as an unsaved draft on the canvas (not deployed).${inferredLineageNote(detectedReads, detectedWrites)}`
+			return `Pipeline node '${path}' updated as an unsaved draft on the canvas (not deployed).${inferredLineageNote(detectedReads, detectedWrites)}${unconfiguredTriggersNote(path, await pipeline.unconfiguredTriggers(path))}`
+		}
+	},
+	{
+		requires: WRITE_DRAFT,
+		def: setPipelineTriggerToolDef,
+		showDetails: true,
+		fn: async ({ args, helpers, toolId, toolCallbacks }) => {
+			const { path, kind, config } = setPipelineTriggerSchema.parse(args)
+			const pipeline = pipelineForPath(helpers, path)
+			toolCallbacks.setToolStatus(toolId, { content: `Setting the ${kind} trigger of '${path}'...` })
+			const valid = validateTriggerConfig(kind, config, path)
+			const { path: triggerPath, replaced } = await pipeline.setNodeTrigger(path, kind, valid)
+			recordPipelineModified(toolCallbacks, pipeline)
+			toolCallbacks.setToolStatus(toolId, {
+				content: `${replaced ? 'Updated' : 'Attached'} ${kind} trigger '${triggerPath}'`,
+				result: 'Success'
+			})
+			return `${replaced ? 'Replaced' : 'Attached'} the ${kind} trigger draft '${triggerPath}' on '${path}'. It deploys with the pipeline.${unconfiguredTriggersNote(path, await pipeline.unconfiguredTriggers(path))}`
 		}
 	},
 	{
@@ -443,7 +548,7 @@ Data Pipeline editor (ACTIVE):
 - ${openLine} A pipeline is a DAG of scripts (nodes) connected by storage assets (DuckLake tables, data tables, S3 objects, volumes, resources) and execution triggers.
 - Annotations are top-of-file comments in the NODE'S OWN comment syntax: \`--\` for SQL (duckdb/postgresql), \`#\` for python3/bash, \`//\` for bun/TS. The \`//\` shown below is the TS form — translate it (a \`// pipeline\` line in a SQL node is a syntax error that won't deploy).
 - A script becomes a pipeline node when its source starts with the \`// pipeline\` annotation. Declare execution-DAG inputs with \`// on <asset-uri | schedule | webhook | email | kafka | mqtt | amqp | nats | postgres | sqs | gcp | data_upload>\` (e.g. \`// on ducklake://main/orders\`). Outputs are inferred from what the body writes (wmill SDK calls / SQL CREATE TABLE / writeS3File); declare a managed output with \`// materialize <asset-uri>\`. Optional badges: \`// partitioned <daily|hourly|weekly|monthly|dynamic>\`, \`// freshness <duration>\`, \`// tag <name>\`, \`// retry <count> [delay]\`, \`// data_test <kind> ...\` (managed DuckLake targets only — deploy rejects it beside a \`dbt://\` target), \`// measure <name> = <agg> [where <pred>]\`, \`// dimension <name> = <expr>\`.
-- \`materialize\` (the managed output): a managed \`// materialize ducklake://<name>/<table>\` means the runtime writes the node's output table FOR you — write the body as a single SELECT and the runtime wraps it in the create/replace, so do NOT also write your own CREATE TABLE / INSERT. The \`dbt://\` target below is the opposite: the node writes its own DDL and none of the write strategies apply to it. IMPORTANT: a MANAGED \`// materialize\` is **DuckDB-only** and its target MUST be a DuckLake table (\`ducklake://<name>/<table>\`) — deploy rejects a \`ducklake://\` target on any other language. For a \`python3\`/\`bun\`/\`postgresql\` node writing the lake, do NOT use \`// materialize\`; write the output via the SDK instead (e.g. \`wmill.writeS3File(...)\`, a \`CREATE TABLE\` in postgresql, or \`wmill.databaseUrlFromResource\`/ducklake helpers) and let the output be inferred. Reach for \`duckdb\` when a node should materialize a DuckLake table. The one target any language BUT DBT'S OWN may declare (a dbt project's writes come from its manifest, so \`// materialize\` on a dbt script is rejected at deploy) is a WAREHOUSE RELATION: \`// materialize manual dbt://<warehouse>/<schema>/<name>\`, with \`<warehouse>\` a warehouse the workspace configures under Settings → dbt. \`manual\` is its only mode — nothing generates warehouse DDL, so the node issues its own write and the annotation records the outcome. Use it on an ingestion node a dbt project reads as a \`source\`: the declared relation and the dbt model become ONE graph node, and a downstream \`// on dbt://<warehouse>/<schema>/<name>\` fires when that node completes. Write strategy: with no option it REPLACES the whole table each run (full refresh; the only mode whose output columns may change); \`// materialize <uri> append\` INSERT-appends rows (incremental); \`// materialize <uri> key=<col>\` merges/upserts on \`<col>\`. \`// materialize manual <uri>\` opts OUT of managed writes — the script writes its own DDL and the annotation only records the output asset for lineage. \`materialize\` is paired with partitioning for incremental pipelines: a \`// partitioned <daily|hourly|weekly|monthly|dynamic>\` node runs once per partition (append/merge into a fixed-schema table), and the \`{partition}\` token — usable in any asset URI AND in the body SQL — is substituted with the current partition's IDENTITY string at run time. To filter the source to the active slice on a time grain, use the runtime-injected macro: \`WHERE wm_partition(<ts_col>) = {partition}\`. \`wm_partition(ts)\` buckets a timestamp with the exact identity format the runtime used (daily/hourly/weekly/monthly), so it always matches and you never hand-write a \`strftime\` format. Do NOT write \`= TIMESTAMP {partition}\`: the identity string is not a valid timestamp literal for hourly/weekly/monthly and errors at runtime. For \`dynamic\` partitioning the identity is your caller-supplied key (not a timestamp, no macro), so filter on it directly: \`WHERE <your_key_col> = {partition}\`. \`materialize\` is an output DECLARATION on the node — it is not a command; there is no "materialize run".
+- \`materialize\` (the managed output): a managed \`// materialize ducklake://<name>/<table>\` means the runtime writes the node's output table FOR you — write the body as a single SELECT and the runtime wraps it in the create/replace, so do NOT also write your own CREATE TABLE / INSERT. The \`dbt://\` target below is the opposite: the node writes its own DDL and none of the write strategies apply to it. IMPORTANT: a MANAGED \`// materialize\` is **DuckDB-only** and its target MUST be a DuckLake table (\`ducklake://<name>/<table>\`) — deploy rejects a \`ducklake://\` target on any other language. For a \`python3\`/\`bun\`/\`postgresql\` node writing the lake, do NOT use \`// materialize\`; write the table straight into DuckLake with \`wmill.ducklake()\` (or a data table with \`wmill.datatable()\`), or a \`CREATE TABLE\` in postgresql and let the output be inferred. Reach for \`duckdb\` when a node should materialize a DuckLake table. The one target any language BUT DBT'S OWN may declare (a dbt project's writes come from its manifest, so \`// materialize\` on a dbt script is rejected at deploy) is a WAREHOUSE RELATION: \`// materialize manual dbt://<warehouse>/<schema>/<name>\`, with \`<warehouse>\` a warehouse the workspace configures under Settings → dbt. \`manual\` is its only mode — nothing generates warehouse DDL, so the node issues its own write and the annotation records the outcome. Use it on an ingestion node a dbt project reads as a \`source\`: the declared relation and the dbt model become ONE graph node, and a downstream \`// on dbt://<warehouse>/<schema>/<name>\` fires when that node completes. Write strategy: with no option it REPLACES the whole table each run (full refresh; the only mode whose output columns may change); \`// materialize <uri> append\` INSERT-appends rows (incremental); \`// materialize <uri> key=<col>\` merges/upserts on \`<col>\`. \`// materialize manual <uri>\` opts OUT of managed writes — the script writes its own DDL and the annotation only records the output asset for lineage. \`materialize\` is paired with partitioning for incremental pipelines: a \`// partitioned <daily|hourly|weekly|monthly|dynamic>\` node runs once per partition (append/merge into a fixed-schema table), and the \`{partition}\` token — usable in any asset URI AND in the body SQL — is substituted with the current partition's IDENTITY string at run time. To filter the source to the active slice on a time grain, use the runtime-injected macro: \`WHERE wm_partition(<ts_col>) = {partition}\`. \`wm_partition(ts)\` buckets a timestamp with the exact identity format the runtime used (daily/hourly/weekly/monthly), so it always matches and you never hand-write a \`strftime\` format. Do NOT write \`= TIMESTAMP {partition}\`: the identity string is not a valid timestamp literal for hourly/weekly/monthly and errors at runtime. For \`dynamic\` partitioning the identity is your caller-supplied key (not a timestamp, no macro), so filter on it directly: \`WHERE <your_key_col> = {partition}\`. \`materialize\` is an output DECLARATION on the node — it is not a command; there is no "materialize run".
 - \`measure\` / \`dimension\` (declared metrics): on a node that materializes a DuckLake table, \`// measure <name> = <aggregate> [where <predicate>]\` names the canonical way to aggregate that table (e.g. \`// measure revenue = sum(amount) where not is_refund\`), and \`// dimension <name> = <expr>\` names a way to slice it (e.g. \`// dimension region = region\`, \`// dimension month = date_trunc('month', ordered_at)\`). They execute nothing: they are catalogued at deploy so the editor and other agents can reuse the definition instead of re-deriving it and silently disagreeing. Keep the predicate in the \`where\` clause rather than folding it into the aggregate: it is rendered as \`<agg> FILTER (WHERE <pred>)\`, which is what lets two measures with different predicates sit under one GROUP BY. DuckLake-only, and only meaningful next to \`// materialize\`. Declare one when a number carries a judgement call someone else would get wrong (refunds excluded, test rows dropped, which column is the amount); do NOT blanket every table with measures, an obvious \`count(*)\` earns nothing. To USE a metric another node declares, read that node with read_pipeline_node and reuse its exact expression rather than guessing it.
 - Use get_pipeline_graph to see the current nodes/assets/triggers, and read_pipeline_node before editing one.
 - ${pathLine}${
@@ -451,6 +556,7 @@ Data Pipeline editor (ACTIVE):
 			? `
 - Build new nodes with build_pipeline_node and edit existing ones with edit_pipeline_node. These apply directly as unsaved drafts on the canvas (like the flow/script editor applies AI edits) — they DO NOT deploy. There is no separate Accept/Reject step. Prefer these over the generic write_script/edit_script draft tools while a pipeline is open.
 - Reuse existing asset paths from the graph when wiring a downstream node to an upstream one (read the upstream's write asset, then \`// on\` that same URI).
+- TRIGGERS: an ingestion node (one with no upstream asset to react to) must declare how it starts, e.g. \`// on schedule\`, and every \`on <schedule|email|kafka|mqtt|amqp|nats|postgres|sqs|gcp>\` line needs its trigger: pass it in build_pipeline_node's \`triggers\` (or call set_pipeline_trigger) in the same step you build the node. A schedule needs a 6-field cron and an IANA timezone — pick a sensible cadence from the request (e.g. daily at 06:00 UTC for a daily sync) rather than leaving it empty; for the other kinds call get_trigger_schema(kind) for the config fields and ask the user for any connection/resource you cannot find. Never leave a declared trigger without one: the node would deploy with nothing to start it. \`webhook\` and \`data_upload\` need no trigger.
 - Only deploy when the user explicitly asks; the user deploys drafts from the canvas.`
 			: ''
 	}`

frontend/src/lib/components/sessions/PipelineEditorView.svelte

@@ -109,7 +109,8 @@
 			liveAnnotations: pe.liveAnnotations,
 			inferredWritesByPath: assetPrefetch.inferredWritesByPath,
 			inferredReadsByPath: assetPrefetch.inferredReadsByPath,
-			annotatedNativeKindsByPath: assetPrefetch.annotatedNativeKindsByPath
+			annotatedNativeKindsByPath: assetPrefetch.annotatedNativeKindsByPath,
+			triggerDrafts: pe.triggerDrafts.values()
 		})
 	)
 
@@ -121,6 +122,8 @@
 		setDrafts: (next) => (pe.drafts = next),
 		newDraftLocalId: pe.newDraftLocalId,
 		hasTriggerDrafts: () => pe.triggerDrafts.size > 0,
+		getTriggerDrafts: () => pe.triggerDrafts,
+		setTriggerDraft: pe.setTriggerDraft,
 		onForgetPath: (p) => {
 			pe.forgetPath(p)
 			pe.discardTriggerDraftsFor(p)

frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte

@@ -498,6 +498,8 @@
 		getWorkspace: () => $workspaceStore,
 		getResolvedGraph: () => graphWithDraft,
 		hasTriggerDrafts: () => pe.triggerDrafts.size > 0,
+		getTriggerDrafts: () => pe.triggerDrafts,
+		setTriggerDraft: pe.setTriggerDraft,
 		getDrafts: () => pe.drafts,
 		setDrafts: (next) => (pe.drafts = next),
 		newDraftLocalId: pe.newDraftLocalId,