feat: flag pipeline scripts nothing starts, with schedule/webhook/remove fixes

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

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

@@ -123,6 +123,11 @@ export function scheduleInsteadOfAsset(
 	return hasDirective(out, onTrigger('schedule')) ? out : addDirective(out, 'on schedule')
 }
 
+/** Adds a `// on <trigger kind>` line (`schedule`, `webhook`, …) unless it has one. */
+export function addTriggerDirective(content: string, kind: string): string {
+	return hasDirective(content, onTrigger(kind)) ? content : addDirective(content, `on ${kind}`)
+}
+
 /** Drops a `// on <trigger kind>` line (`kafka`, `schedule`, …). */
 export function removeTriggerDirective(content: string, kind: string): string {
 	return removeDirective(content, onTrigger(kind))

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

@@ -1275,6 +1275,25 @@ describe('pipelineNodeErrors', () => {
 		expect([...errors.scripts.keys()]).toEqual(['f/x/cons'])
 	})
 
+	it('flags a pipeline script nothing starts, but not a macro library or a triggered one', () => {
+		const g = baseGraph({
+			runnables: [
+				{ path: 'f/x/stray', usage_kind: 'script', in_pipeline: true },
+				{ path: 'f/x/lib', usage_kind: 'script', in_pipeline: true, macros: [{ name: 'm' }] },
+				{ path: 'f/x/timed', usage_kind: 'script', in_pipeline: true }
+			] as any,
+			triggers: [
+				{
+					trigger_kind: 'schedule',
+					path: 'f/x/timed_schedule',
+					runnable_kind: 'script',
+					runnable_path: 'f/x/timed'
+				}
+			]
+		})
+		expect([...pipelineNodeErrors(g, new Map()).untriggered]).toEqual(['f/x/stray'])
+	})
+
 	it('clears a subscription the workspace writes outside the pipeline', () => {
 		const errors = pipelineNodeErrors(
 			graph,

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

@@ -327,6 +327,8 @@ export type PipelineNodeErrors = {
 	unwritten: Map<string, { asset: { kind: AssetKind; path: string }; subscribers: string[] }>
 	/** Script path → the unwritten asset it runs on (its first, when several). */
 	waitsOn: Map<string, { kind: AssetKind; path: string }>
+	/** Pipeline scripts nothing starts: no trigger of any kind. */
+	untriggered: Set<string>
 }
 
 /**
@@ -384,7 +386,17 @@ export function pipelineNodeErrors(
 			}
 		}
 	}
-	return { assets: errors, scripts: scriptErrors, unwritten, waitsOn }
+	// A pipeline script nothing starts runs only by hand. dbt projects and macro
+	// libraries are never started on their own, so they are left out.
+	const triggered = new Set(graph.triggers.map((t) => t.runnable_path))
+	const untriggered = new Set<string>()
+	for (const r of graph.runnables) {
+		if (r.usage_kind !== 'script' || !r.in_pipeline || r.dbt || r.macros?.length) continue
+		if (triggered.has(r.path) || scriptErrors.has(r.path)) continue
+		untriggered.add(r.path)
+		scriptErrors.set(r.path, 'nothing starts it: it has no trigger')
+	}
+	return { assets: errors, scripts: scriptErrors, unwritten, waitsOn, untriggered }
 }
 
 function referencedAssetKeys(

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

@@ -101,6 +101,7 @@
 	import type { NodeFixSpec } from '$lib/components/assets/AssetGraph/NodeFixButton.svelte'
 	import { TRIGGER_NODE_STYLE } from '$lib/components/assets/AssetGraph/TriggerNode.svelte'
 	import {
+		addTriggerDirective,
 		assetRef,
 		listenToAsset,
 		removeTriggerDirective,
@@ -1476,6 +1477,55 @@
 				]
 			})
 		}
+		for (const path of nodeErrors.untriggered) {
+			const unsaved = displayGraph.runnables.some((r) => r.path === path && r.unsaved)
+			fixes.set(`script:${path}`, {
+				explainer: [
+					'Nothing starts ',
+					scriptLink(path),
+					': it has no trigger, so it only runs by hand.'
+				],
+				actions: [
+					{
+						label: 'Run it on a schedule',
+						detail: `Opens the schedule setup. On create: adds \`on schedule\` in ${path}, with the schedule as a draft.`,
+						scheduleFor: {
+							script: path,
+							onSchedule: async (schedule) => {
+								const ok = await editScriptHeader(
+									path,
+									(c) => addTriggerDirective(c, 'schedule'),
+									'Added a schedule'
+								)
+								if (!ok) return
+								if (!pe.setTriggerDraft({ kind: 'schedule', config: { ...schedule, script_path: path } })) {
+									sendUserToast(
+										`Another draft schedule already uses the path ${schedule.path}; ${path} has no schedule yet`,
+										true
+									)
+								}
+							}
+						}
+					},
+					{
+						label: 'Run it from a webhook',
+						detail: `Adds \`on webhook\` in ${path}: it runs when its webhook URL is called. Saved as a draft.`,
+						run: () =>
+							editScriptHeader(path, (c) => addTriggerDirective(c, 'webhook'), 'Added a webhook')
+					},
+					{
+						label: unsaved ? 'Discard it' : 'Remove it',
+						detail: unsaved
+							? 'Drops the draft.'
+							: 'Opens its archive or delete confirmation.',
+						run: () => {
+							if (mode !== 'edit') setMode('edit')
+							handleRunnableMenuRemove({ runnable_kind: 'script', path, unsaved })
+						}
+					}
+				]
+			})
+		}
 		for (const t of displayGraph.triggers) {
 			if (t.trigger_kind === 'asset' || !t.missing) continue
 			if (t.trigger_kind === 'webhook' || t.trigger_kind === 'data_upload') continue