fix: pipeline trigger lookups skip same-path flows; a failed trigger never hides its built node

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-10-01 14:27:26 +02:00
co-authored by Claude Opus 5.5
parent 46ec2995cc
commit 77d58bd7d7
4 changed files with 106 additions and 16 deletions
@@ -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' })
})
})
@@ -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 &&
@@ -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) => ({
@@ -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<PipelineToolHelpers>[] = [
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<PipelineToolHelpers>[] = [
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))}`
}
},