diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts index ef017f0393..713ef8f6b2 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.test.ts @@ -8,7 +8,8 @@ import type { AssetGraphResponse } from './types' function makeHandle( initial: Array<[string, PipelineDraft]> = [], runnables: Array<{ path: string }> = [], - folder = 'x' + folder = 'x', + triggers: unknown[] = [] ) { let drafts = new Map(initial) let forgotten: string[] = [] @@ -16,7 +17,7 @@ function makeHandle( getFolder: () => folder, getWorkspace: () => 'w', getResolvedGraph: () => - ({ assets: [], runnables, edges: [], triggers: [] }) as unknown as AssetGraphResponse, + ({ assets: [], runnables, edges: [], triggers }) as unknown as AssetGraphResponse, getDrafts: () => drafts, setDrafts: (next) => (drafts = next), newDraftLocalId: () => 'id', @@ -207,4 +208,24 @@ describe('pipeline AI direct-draft helpers', () => { expect(d?.script.tag).toBe('custom') expect(d?.script.content).toBe('-- pipeline\nSELECT 2') }) + + it("does not take a flow's schedule at the same path for the script's", async () => { + const flowSchedule = { + trigger_kind: 'schedule', + runnable_kind: 'flow', + runnable_path: 'f/x/ingest', + path: 'f/x/flow_schedule' + } + const content = '# pipeline\n# on schedule\ndef main():\n return 1' + const { handle } = makeHandle( + [['f/x/ingest', draft({ script: { content, language: 'python3' } as any })]], + [], + 'x', + [flowSchedule] + ) + expect(await handle.unconfiguredTriggers('f/x/ingest')).toEqual(['schedule']) + await expect( + handle.setNodeTrigger('f/x/ingest', 'schedule', { schedule: '0 0 6 * * *', timezone: 'UTC' }) + ).resolves.toMatchObject({ path: 'f/x/ingest_schedule' }) + }) }) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts index e360a6b577..f84848ee95 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts @@ -240,7 +240,11 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC .getResolvedGraph() .triggers.filter( (t) => - t.trigger_kind !== 'asset' && t.runnable_path === path && t.path && !t.missing + t.trigger_kind !== 'asset' && + t.runnable_kind === 'script' && + t.runnable_path === path && + t.path && + !t.missing ) .map((t) => t.trigger_kind) ) @@ -406,6 +410,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC .triggers.find( (t) => t.trigger_kind === kind && + t.runnable_kind === 'script' && t.runnable_path === path && t.path && !t.missing && 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 57f68df10b..6c4539ef85 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts @@ -176,6 +176,54 @@ describe('pipeline tools', () => { expect(calls.proposeNode).toBeUndefined() }) + it('build_pipeline_node keeps and records the node when a trigger fails to attach', async () => { + const { helpers, calls } = makeHelpers({ + setNodeTrigger: async () => { + throw new Error('path taken') + } + }) + const modified: string[] = [] + const out = await toolByName('build_pipeline_node').fn({ + args: { + path: 'f/analytics/ingest', + language: 'python3', + content: '# pipeline\n# on schedule\ndef main():\n return 1', + summary: 'Ingest', + triggers: [{ kind: 'schedule', config: { schedule: '0 0 6 * * *', timezone: 'UTC' } }] + }, + workspace: 'w', + helpers, + toolCallbacks: { + ...noopCallbacks(), + onItemModified: (kind, path) => modified.push(`${kind}:${path}`) + }, + toolId: 't' + }) + expect(calls.proposeNode).toHaveLength(1) + expect(modified).toEqual(['data_pipeline:f/analytics/data_pipeline']) + expect(out).toContain('path taken') + }) + + it('build_pipeline_node refuses a trigger the content does not declare, before staging', async () => { + const { helpers, calls } = makeHelpers() + await expect( + toolByName('build_pipeline_node').fn({ + args: { + path: 'f/analytics/ingest', + language: 'python3', + content: '# pipeline\ndef main():\n return 1', + summary: 'Ingest', + triggers: [{ kind: 'schedule', config: { schedule: '0 0 6 * * *', timezone: 'UTC' } }] + }, + workspace: 'w', + helpers, + toolCallbacks: noopCallbacks(), + toolId: 't' + }) + ).rejects.toThrow(/on 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 f47decce08..8d71b00279 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.ts @@ -15,6 +15,7 @@ import type { PipelineTriggerDraftKind } from '$lib/components/assets/AssetGraph import { DRAFTABLE_TRIGGER_KINDS } from '$lib/components/assets/AssetGraph/pipelineTriggerDrafts' import { scheduleRequestSchema, triggerRequestSchemas } from '../workspaceToolsZod.gen' import { normalizePipelineFolder } from '$lib/utils/pipelineFolder' +import { parsePipelineAnnotations } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations' // ============================================================================ // Pipeline AI chat tools. @@ -437,11 +438,16 @@ export const pipelineTools: SessionTool[] = [ const { path, language, content, summary, output_kind, triggers } = buildPipelineNodeSchema.parse(args) const pipeline = await 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) - })) + // Checked before the node is staged, so a bad trigger leaves nothing half-built. + const declared = new Set(parsePipelineAnnotations(content).nativeTriggers.map((n) => n.kind)) + const configs = (triggers ?? []).map((t) => { + if (!declared.has(t.kind)) { + throw new Error( + `The content declares no \`on ${t.kind}\`, so a ${t.kind} trigger has nothing to bind to. Add the annotation line or drop that trigger. Nothing was built.` + ) + } + return { kind: t.kind, config: validateTriggerConfig(t.kind, t.config, path) } + }) toolCallbacks.setToolStatus(toolId, { content: `Building node '${path}'...` }) const { detectedReads, detectedWrites } = await pipeline.proposeNode({ path, @@ -450,19 +456,29 @@ export const pipelineTools: SessionTool[] = [ summary, 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}'`) - } + // Recorded as soon as it is staged: a trigger that fails below leaves the node. recordPipelineModified(toolCallbacks, pipeline) + const attached: string[] = [] + const failed: string[] = [] + for (const t of configs) { + try { + const { path: triggerPath } = await pipeline.setNodeTrigger(path, t.kind, t.config) + attached.push(`${t.kind} '${triggerPath}'`) + } catch (e) { + failed.push(`${t.kind}: ${e instanceof Error ? e.message : String(e)}`) + } + } toolCallbacks.setToolStatus(toolId, { content: `Added draft node '${path}'`, result: 'Success' }) - const attachedNote = attached.length - ? ` Attached trigger draft${attached.length > 1 ? 's' : ''}: ${attached.join(', ')}.` - : '' + const attachedNote = + (attached.length + ? ` Attached trigger draft${attached.length > 1 ? 's' : ''}: ${attached.join(', ')}.` + : '') + + (failed.length + ? ` The node was built, but these triggers were not attached — fix them with set_pipeline_trigger: ${failed.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))}` } },