From 28cb6704f3df5a20ee3ae6a7242aa3a3c2a684ae Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 1 Oct 2026 13:54:19 +0200 Subject: [PATCH] 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) --- .../AssetGraph/pipelineAiHelpers.test.ts | 28 ++++- .../assets/AssetGraph/pipelineAiHelpers.ts | 97 ++++++++++++++- .../copilot/chat/AIChatManager.test.ts | 2 + .../copilot/chat/pipeline/core.test.ts | 49 ++++++++ .../components/copilot/chat/pipeline/core.ts | 116 +++++++++++++++++- .../sessions/PipelineEditorView.svelte | 5 +- .../(logged)/pipeline/[folder]/+page.svelte | 2 + 7 files changed, 288 insertions(+), 11 deletions(-) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts index d1d5c1b233..ef017f0393 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts +++ b/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/) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts index bde960b06d..9f2cb4359f 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts +++ b/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 + /** 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 ` 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.`) diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index 215858bd4e..58b68d051c 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/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 } } diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts index c545fab85a..f8918d792a 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts @@ -66,6 +66,11 @@ function makeHelpers(overrides: Partial = {}): { 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) => ({ diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.ts index c48517335e..cdd7e79eac 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.ts +++ b/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 + /** Trigger kinds the node declares (`on `) that no trigger row or draft fills. */ + unconfiguredTriggers: (path: string) => Promise + /** 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 + ) => Promise<{ path: string; replaced: boolean }> /** Preview-run a node (draft body preferred). Returns the started job id. */ testNode: (path: string, args?: Record) => Promise } @@ -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 `.'), + 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 ` (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, + nodePath: string +): Record { + 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('.') || ''}: ${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('.') || ''}: ${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 `), 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 ` 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[] = [ 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[] = [ 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[] = [ 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 \` (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 \`. Optional badges: \`// partitioned \`, \`// freshness \`, \`// tag \`, \`// retry [delay]\`, \`// data_test ...\` (managed DuckLake targets only — deploy rejects it beside a \`dbt://\` target), \`// measure = [where ]\`, \`// dimension = \`. -- \`materialize\` (the managed output): a managed \`// materialize ducklake:///\` 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:///
\`) — 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:////\`, with \`\` 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:////\` 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 append\` INSERT-appends rows (incremental); \`// materialize key=\` merges/upserts on \`\`. \`// materialize manual \` 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 \` 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() = {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 = {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:///
\` 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:///
\`) — 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:////\`, with \`\` 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:////\` 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 append\` INSERT-appends rows (incremental); \`// materialize key=\` merges/upserts on \`\`. \`// materialize manual \` 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 \` 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() = {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 = {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 = [where ]\` names the canonical way to aggregate that table (e.g. \`// measure revenue = sum(amount) where not is_refund\`), and \`// dimension = \` 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 \` FILTER (WHERE )\`, 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 \` 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.` : '' }` diff --git a/frontend/src/lib/components/sessions/PipelineEditorView.svelte b/frontend/src/lib/components/sessions/PipelineEditorView.svelte index 6cca4ce2e0..a1936791dd 100644 --- a/frontend/src/lib/components/sessions/PipelineEditorView.svelte +++ b/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) diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 88b311c7f7..7cfc83a4aa 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/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,