fix: route long pipeline edges around nodes in the target column, and flag scripts waiting on unproduced assets

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

frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte

@@ -531,6 +531,7 @@
 					path: r.path,
 					summary: r.summary,
 					language: r.language,
+					error: r.error,
 					in_pipeline: r.in_pipeline ?? false,
 					partition_kind: r.partition_kind,
 					freshness: r.freshness,
@@ -1070,11 +1071,12 @@
 			.filter((n) => n.id !== ADD_NODE_ID)
 			.map((n) => ({ id: n.id, cx: n.position.x + NODE.width / 2, cy: n.position.y }))
 	)
-	// Detour lane for an edge whose straight run would pass over an unrelated
-	// node (the failure the same-column gutter can't see). For each node
-	// strictly between the endpoints' rows, sample the straight line at that
-	// row; if the node sits under it, route around the obstacle on the side the
-	// edge is already heading. Returns the outermost lane x clearing every
+	// Detour lane for an edge whose run would pass over an unrelated node. An
+	// edge spanning rows bends inside the first gap below its source and then
+	// runs straight down the *target's* column (see AssetGraphEdge), so every
+	// node strictly between the endpoints' rows is tested against the target's
+	// x, not against a source→target line. A crossed node is routed around on
+	// the side the target lies. Returns the outermost lane x clearing every
 	// crossed node, or undefined when the corridor is clear. O(nodes) per edge.
 	const HALF_W = NODE.width / 2
 	const ROUTE_PAD = NODE.gap.horizontal / 2
@@ -1088,8 +1090,7 @@
 			if (n.id === sourceId || n.id === targetId) continue
 			// strictly between the two rows
 			if ((n.cy - s.cy) / dyTot <= 0.01 || (n.cy - s.cy) / dyTot >= 0.99) continue
-			const edgeX = s.cx + (t.cx - s.cx) * ((n.cy - s.cy) / dyTot)
-			if (Math.abs(edgeX - n.cx) >= HALF_W + 8) continue
+			if (Math.abs(t.cx - n.cx) >= HALF_W + 8) continue
 			// Crossed: a lane just outside this node, toward the target side.
 			const side = t.cx >= n.cx ? 1 : -1
 			const candidate = n.cx + side * (HALF_W + ROUTE_PAD)

frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte

@@ -41,6 +41,8 @@
 			path: string
 			summary?: string
 			language?: ScriptLang
+			// Why the script is misconfigured, when it is: renders it red.
+			error?: string
 			in_pipeline?: boolean
 			partition_kind?: 'daily' | 'hourly' | 'weekly' | 'monthly' | 'dynamic'
 			freshness?: string
@@ -226,10 +228,10 @@
 		surface="primary"
 		kindLabel="{kindLabel}{data.unsaved ? ' · draft' : ''}"
 		title={data.summary || data.path}
-		tooltip={nodeTooltip}
+		tooltip={data.error ? `${data.path}: ${data.error}` : nodeTooltip}
 		{selected}
 		draft={data.unsaved}
-		tone={computingNow ? 'running' : undefined}
+		tone={computingNow ? 'running' : data.error ? 'error' : undefined}
 	>
 		{#snippet icon()}
 			{#if data.language && data.runnable_kind === 'script'}

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

@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
 import {
 	resolveGraph,
 	computeMutedReadKeys,
-	pipelineErrorAssetKeys,
+	pipelineNodeErrors,
 	dbtAssociations,
 	type ResolveGraphInput
 } from './resolveGraph'
@@ -1233,7 +1233,7 @@ describe('dbtAssociations with two writers of one relation', () => {
 	})
 })
 
-describe('pipelineErrorAssetKeys', () => {
+describe('pipelineNodeErrors', () => {
 	const edge = (runnable_path: string, asset_path: string, access_type: 'r' | 'w') => ({
 		runnable_path,
 		runnable_kind: 'script' as const,
@@ -1258,13 +1258,20 @@ describe('pipelineErrorAssetKeys', () => {
 		]
 	})
 
-	it('flags an unused asset, and a `// on` with no producer, but not a plain read', () => {
-		const onRefs = (...paths: string[]) => paths.map((path) => ({ kind: 'ducklake' as const, path }))
-		expect([...pipelineErrorAssetKeys(graph, new Map()).keys()]).toEqual(['ducklake:main/orphan'])
-		const errors = pipelineErrorAssetKeys(
+	it('flags an unused asset, and a `// on` with no producer and its script, but not a plain read', () => {
+		const onRefs = (...paths: string[]) =>
+			paths.map((path) => ({ kind: 'ducklake' as const, path }))
+		const none = pipelineNodeErrors(graph, new Map())
+		expect([...none.assets.keys()]).toEqual(['ducklake:main/orphan'])
+		expect(none.scripts.size).toBe(0)
+		const errors = pipelineNodeErrors(
 			graph,
 			new Map([['f/x/cons', onRefs('main/produced', 'main/external')]])
 		)
-		expect([...errors.keys()].sort()).toEqual(['ducklake:main/external', 'ducklake:main/orphan'])
+		expect([...errors.assets.keys()].sort()).toEqual([
+			'ducklake:main/external',
+			'ducklake:main/orphan'
+		])
+		expect([...errors.scripts.keys()]).toEqual(['f/x/cons'])
 	})
 })

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

@@ -314,16 +314,17 @@ function withTriggerDrafts(
 }
 
 /**
- * Assets the pipeline cannot work with, keyed `kind:path`, with the reason: one
- * nothing uses at all, and one a script subscribes to (`// on <asset>`) that no
- * runnable in the graph writes — that subscription can never fire. A body read of
- * an external asset is fine and is not passed here: only the explicit `// on`
+ * Nodes the pipeline cannot work with, with the reason. Assets (keyed
+ * `kind:path`): one nothing uses at all, and one a script subscribes to
+ * (`// on <asset>`) that no runnable in the graph writes. Scripts (keyed by
+ * path): the subscriber, whose subscription can never fire. A body read of an
+ * external asset is fine and is not passed here: only the explicit `// on`
  * annotations, per script path, count as subscriptions.
  */
-export function pipelineErrorAssetKeys(
+export function pipelineNodeErrors(
 	graph: AssetGraphResponse,
 	explicitOnByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
-): Map<string, string> {
+): { assets: Map<string, string>; scripts: Map<string, string> } {
 	const referenced = referencedAssetKeys(graph)
 	const written = new Set(
 		graph.edges
@@ -332,6 +333,7 @@ export function pipelineErrorAssetKeys(
 	)
 	for (const e of graph.dbt_edges ?? []) written.add(`dbt:${e.from_asset_path}`)
 	const errors = new Map<string, string>()
+	const scriptErrors = new Map<string, string>()
 	for (const a of graph.assets) {
 		const key = `${a.kind}:${a.path}`
 		if (!referenced.has(key)) errors.set(key, 'nothing in the pipeline uses it')
@@ -343,12 +345,16 @@ export function pipelineErrorAssetKeys(
 		if (!scripts.has(path)) continue
 		for (const ref of refs) {
 			const key = `${ref.kind}:${ref.path}`
-			if (!written.has(key) && !errors.has(key)) {
+			if (written.has(key)) continue
+			if (!errors.has(key)) {
 				errors.set(key, `${path} runs on it, but nothing in the pipeline writes it`)
 			}
+			if (!scriptErrors.has(path)) {
+				scriptErrors.set(path, `runs on ${ref.path}, which nothing in the pipeline writes`)
+			}
 		}
 	}
-	return errors
+	return { assets: errors, scripts: scriptErrors }
 }
 
 function referencedAssetKeys(

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

@@ -6,7 +6,7 @@ export type GraphUsageKind = 'script' | 'flow'
 export interface AssetGraphAssetNode {
 	kind: AssetKind
 	path: string
-	// Filled in client-side: the node is misconfigured (see `pipelineErrorAssetKeys`).
+	// Filled in client-side: the node is misconfigured (see `pipelineNodeErrors`).
 	error?: string
 	// Fork workspaces only: 'fork' when this ducklake asset was materialized in
 	// the fork itself, 'deferred' when reads fall back to the parent workspace's
@@ -76,6 +76,8 @@ export interface AssetGraphRunnableNode {
 	// node with the summary over the path, and shows the language's icon.
 	summary?: string
 	language?: import('$lib/gen').ScriptLang
+	// Why the script is misconfigured, when it is (see `pipelineNodeErrors`).
+	error?: string
 	// Script has `// pipeline` annotation. Drives the pipeline-member
 	// visual state; unrelated to what the script actually writes (that's
 	// parsed separately into lineage edges).

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

@@ -44,7 +44,7 @@
 	} from '$lib/components/assets/AssetGraph/columnLineageGraph'
 	import { useDbtColumnLineage } from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte'
 	import {
-		pipelineErrorAssetKeys,
+		pipelineNodeErrors,
 		resolveGraph
 	} from '$lib/components/assets/AssetGraph/resolveGraph'
 	import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
@@ -1365,14 +1365,14 @@
 					: []
 			if (refs.length > 0) explicitOnByPath.set(r.path, refs)
 		}
-		const errors = pipelineErrorAssetKeys(g, explicitOnByPath)
+		const errors = pipelineNodeErrors(g, explicitOnByPath)
 		return {
 			...g,
 			assets:
-				errors.size === 0
+				errors.assets.size === 0
 					? g.assets
 					: g.assets.map((a) => {
-							const error = errors.get(`${a.kind}:${a.path}`)
+							const error = errors.assets.get(`${a.kind}:${a.path}`)
 							return error ? { ...a, error } : a
 						}),
 			runnables: g.runnables.map((r) => {
@@ -1380,7 +1380,8 @@
 				const meta = scriptMetaByPath.get(r.path)
 				const summary = draft?.summary || meta?.summary
 				const language = draft?.language ?? meta?.language
-				return summary || language ? { ...r, summary, language } : r
+				const error = r.usage_kind === 'script' ? errors.scripts.get(r.path) : undefined
+				return summary || language || error ? { ...r, summary, language, error } : r
 			})
 		}
 	}