fix(pipelines): live materialize/dataset edits reflect on the graph; no phantom draft after deploy

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -482,6 +482,7 @@
 	// restores its input. Guarded on the path so a staging round-trip
 	// (emit → page → runFormInitialArgs) doesn't re-seed and loop. The read-only
 	// branch uses PipelineScriptView's own onArgsChange instead.
+	let args = $state<Record<string, any>>({})
 	let argsSeedPath: string | undefined = undefined
 	$effect.pre(() => {
 		const p = script?.path
@@ -532,6 +533,16 @@
 				const orig = latest && latest.path === captured.path ? latest : origAtRegister
 				if (!orig || orig.path !== captured.path) return
 				if ((captured.content ?? '') === (orig.content ?? '')) return
+				// The effect can re-run mid-save — after createScript resolved
+				// but before the refetch lands — with `orig` still holding the
+				// pre-deploy version. The buffer isn't "unsaved edits" then, it
+				// IS the new deployed head; emitting would resurrect it as a
+				// phantom draft identical to what was just deployed.
+				if (
+					deployedFromPane?.path === captured.path &&
+					(captured.content ?? '') === deployedFromPane.content
+				)
+					return
 			}
 			const writes = (liveBodyAssets ?? [])
 				.filter((a) => {
@@ -547,8 +558,11 @@
 		}
 	})
 
-	let args = $state<Record<string, any>>({})
 	let saving = $state(false)
+	// What this pane last deployed, so the persist-back cleanup can tell "the
+	// buffer equals the new deployed head" from real unsaved edits (plain
+	// variable: only read inside the untracked cleanup).
+	let deployedFromPane: { path: string; content: string } | undefined = undefined
 	let isDraft = $derived(draftScript != undefined)
 
 	// Single trash-bin button opens one modal that exposes both Archive
@@ -714,6 +728,7 @@
 			// backend rejects as a lineage fork ("no 2 scripts can have the
 			// same parent").
 			if (typeof newHash === 'string' && newHash) script.hash = newHash
+			deployedFromPane = { path: script.path, content: script.content ?? '' }
 			sendUserToast(`Saved ${script.path}`)
 			// Authoritative save-time schema-contract check (pipelines gap #2b):
 			// warn-only, post-commit. Fire-and-forget — must never gate the save.

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

@@ -796,3 +796,155 @@ describe('computeMutedReadKeys', () => {
 		expect(computeMutedReadKeys([readEdge('ducklake', 'main.orders')], [], flow).size).toBe(0)
 	})
 })
+
+describe('live buffer overlays (open script)', () => {
+	// A deployed producer: `f/x/prod` materializes ducklake main/orders.
+	const deployedProducer = () =>
+		baseGraph({
+			assets: [{ kind: 'ducklake', path: 'main/orders' }],
+			runnables: [{ path: 'f/x/prod', usage_kind: 'script', in_pipeline: true }],
+			edges: [
+				{
+					runnable_path: 'f/x/prod',
+					runnable_kind: 'script',
+					asset_kind: 'ducklake',
+					asset_path: 'main/orders',
+					access_type: 'w'
+				}
+			]
+		})
+
+	it('open draft: live annotations win over the stale draft snapshot for the materialize target', () => {
+		const drafts = new Map([
+			['f/x/d', { script: { content: '-- materialize ducklake://main/old_target\nselect 1' } }]
+		])
+		const r = resolveGraph(
+			input({
+				drafts,
+				liveBodyAssets: { scriptPath: 'f/x/d', assets: [] },
+				liveAnnotations: {
+					scriptPath: 'f/x/d',
+					annotations: ann({
+						inPipeline: true,
+						materialize: { targetKind: 'ducklake', targetPath: 'main/new_target' }
+					})
+				}
+			})
+		)
+		expect(r.assets).toContainEqual({ kind: 'ducklake', path: 'main/new_target' })
+		expect(r.assets).not.toContainEqual({ kind: 'ducklake', path: 'main/old_target' })
+		expect(r.edges).toContainEqual({
+			runnable_path: 'f/x/d',
+			runnable_kind: 'script',
+			asset_kind: 'ducklake',
+			asset_path: 'main/new_target',
+			access_type: 'w',
+			unsaved: true
+		})
+	})
+
+	it('open saved script: retargeting `// materialize` swaps the write edge live', () => {
+		const r = resolveGraph(
+			input({
+				base: deployedProducer(),
+				liveBodyAssets: { scriptPath: 'f/x/prod', assets: [] },
+				liveAnnotations: {
+					scriptPath: 'f/x/prod',
+					annotations: ann({
+						inPipeline: true,
+						materialize: { targetKind: 'ducklake', targetPath: 'main/orders_gold' }
+					})
+				}
+			})
+		)
+		// New target surfaces as an unsaved write edge…
+		expect(r.assets).toContainEqual({ kind: 'ducklake', path: 'main/orders_gold' })
+		expect(r.edges).toContainEqual({
+			runnable_path: 'f/x/prod',
+			runnable_kind: 'script',
+			asset_kind: 'ducklake',
+			asset_path: 'main/orders_gold',
+			access_type: 'w',
+			unsaved: true
+		})
+		// …the stale write edge is dropped, but the deployed dataset node stays.
+		expect(
+			r.edges.filter((e) => e.asset_path === 'main/orders' && e.runnable_path === 'f/x/prod')
+		).toEqual([])
+		expect(r.assets).toContainEqual({ kind: 'ducklake', path: 'main/orders' })
+	})
+
+	it('open saved script: unchanged materialize target dedups against the persisted write edge', () => {
+		const r = resolveGraph(
+			input({
+				base: deployedProducer(),
+				liveBodyAssets: { scriptPath: 'f/x/prod', assets: [] },
+				liveAnnotations: {
+					scriptPath: 'f/x/prod',
+					annotations: ann({
+						inPipeline: true,
+						materialize: { targetKind: 'ducklake', targetPath: 'main/orders' }
+					})
+				}
+			})
+		)
+		expect(
+			r.edges.filter(
+				(e) =>
+					e.runnable_path === 'f/x/prod' &&
+					e.asset_path === 'main/orders' &&
+					(e.access_type === 'w' || e.access_type === 'rw')
+			)
+		).toHaveLength(1)
+	})
+
+	it('open saved script: live body reads/writes overlay as unsaved lineage', () => {
+		const base = baseGraph({
+			assets: [{ kind: 'ducklake', path: 'main/orders' }],
+			runnables: [{ path: 'f/x/cons', usage_kind: 'script', in_pipeline: true }],
+			edges: [
+				{
+					runnable_path: 'f/x/cons',
+					runnable_kind: 'script',
+					asset_kind: 'ducklake',
+					asset_path: 'main/orders',
+					access_type: 'r'
+				}
+			]
+		})
+		const r = resolveGraph(
+			input({
+				base,
+				liveBodyAssets: {
+					scriptPath: 'f/x/cons',
+					assets: [duck('main/orders_eu', 'r'), s3('/report.parquet', 'w')]
+				},
+				liveAnnotations: {
+					scriptPath: 'f/x/cons',
+					annotations: ann({ inPipeline: true })
+				}
+			})
+		)
+		// New read + write from the buffer…
+		expect(r.edges).toContainEqual({
+			runnable_path: 'f/x/cons',
+			runnable_kind: 'script',
+			asset_kind: 'ducklake',
+			asset_path: 'main/orders_eu',
+			access_type: 'r',
+			unsaved: true
+		})
+		expect(r.edges).toContainEqual({
+			runnable_path: 'f/x/cons',
+			runnable_kind: 'script',
+			asset_kind: 's3object',
+			asset_path: '/report.parquet',
+			access_type: 'w',
+			unsaved: true
+		})
+		// …the no-longer-read persisted edge is dropped.
+		expect(
+			r.edges.filter((e) => e.asset_path === 'main/orders' && e.runnable_path === 'f/x/cons')
+		).toEqual([])
+	})
+})

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

@@ -408,6 +408,95 @@ function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumul
 	}
 }
 
+/**
+ * The write output(s) a `// materialize <asset>` annotation declares: the
+ * target itself, plus — for a managed scd2 materialize — the `<dim>_current`
+ * companion view (mirrors the deploy path), so a consumer of only the view
+ * links back to this producer instead of orphaning.
+ */
+function materializeOuts(
+	parsed: PipelineAnnotations
+): Array<{ kind: AssetKind; path: string; derivedFrom?: string }> {
+	if (!parsed.materialize) return []
+	const outs: Array<{ kind: AssetKind; path: string; derivedFrom?: string }> = [
+		{ kind: parsed.materialize.targetKind, path: parsed.materialize.targetPath }
+	]
+	const currentPath = scd2CurrentTargetPath(parsed.materialize)
+	if (currentPath) {
+		outs.push({
+			kind: parsed.materialize.targetKind,
+			path: currentPath,
+			derivedFrom: parsed.materialize.targetPath
+		})
+	}
+	return outs
+}
+
+/** Add an output asset node + unsaved write edge for `runnablePath`, deduped
+ *  against the assets/edges already accumulated (base survivors included). */
+function pushWriteOut(
+	acc: Accumulator,
+	runnablePath: string,
+	out: { kind: AssetKind; path: string; derivedFrom?: string }
+) {
+	const existing = acc.assets.find((a) => a.kind === out.kind && a.path === out.path)
+	if (!existing) {
+		acc.assets.push({
+			kind: out.kind,
+			path: out.path,
+			...(out.derivedFrom ? { derived_from: out.derivedFrom } : {})
+		})
+	} else if (out.derivedFrom && existing.derived_from == undefined) {
+		existing.derived_from = out.derivedFrom
+	}
+	const hasWriteEdge = acc.edges.some(
+		(e) =>
+			e.runnable_kind === 'script' &&
+			e.runnable_path === runnablePath &&
+			e.asset_kind === out.kind &&
+			e.asset_path === out.path &&
+			(e.access_type === 'w' || e.access_type === 'rw')
+	)
+	if (hasWriteEdge) return
+	acc.edges.push({
+		runnable_path: runnablePath,
+		runnable_kind: 'script',
+		asset_kind: out.kind,
+		asset_path: out.path,
+		access_type: 'w',
+		unsaved: true
+	})
+}
+
+/** Add an input asset node + unsaved read edge for `runnablePath`, deduped
+ *  against the assets/edges already accumulated. */
+function pushReadIn(
+	acc: Accumulator,
+	runnablePath: string,
+	inp: { kind: AssetKind; path: string }
+) {
+	if (!acc.assets.some((a) => a.kind === inp.kind && a.path === inp.path)) {
+		acc.assets.push({ kind: inp.kind, path: inp.path })
+	}
+	const hasReadEdge = acc.edges.some(
+		(e) =>
+			e.runnable_kind === 'script' &&
+			e.runnable_path === runnablePath &&
+			e.asset_kind === inp.kind &&
+			e.asset_path === inp.path &&
+			(e.access_type === 'r' || e.access_type === 'rw')
+	)
+	if (hasReadEdge) return
+	acc.edges.push({
+		runnable_path: runnablePath,
+		runnable_kind: 'script',
+		asset_kind: inp.kind,
+		asset_path: inp.path,
+		access_type: 'r',
+		unsaved: true
+	})
+}
+
 /**
  * Every draft contributes: a runnable, output asset(s), a write edge, live
  * read lineage for the active draft, plus its seeded asset/native triggers
@@ -416,10 +505,17 @@ function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumul
  */
 function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
 	const { base, drafts, liveBodyAssets, inferredReadsByPath, inferredWritesByPath } = input
-	const { runnables, assets, edges, extraTriggers } = acc
+	const { runnables, assets, extraTriggers } = acc
 
 	for (const [path, d] of drafts) {
-		const parsed = parsePipelineAnnotations(d.script.content)
+		// The open draft's annotations come from the live buffer, not the draft
+		// snapshot — the snapshot only syncs on pane teardown, so parsing it here
+		// would pin annotation-derived outputs (`// materialize` target, badges)
+		// to their pre-edit values until the user clicks away.
+		const parsed =
+			path === input.liveAnnotations.scriptPath
+				? input.liveAnnotations.annotations
+				: parsePipelineAnnotations(d.script.content)
 		// For the open script, fold in the WASM-inferred column lineage (DuckDB
 		// SQL AST) under the same annotation-wins precedence the backend applies
 		// on deploy, so the live preview matches what deploys. Only the open
@@ -490,57 +586,11 @@ function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
 			writeOuts.push(...d.outputAssets)
 		}
 		// `// materialize <asset>` declares a write output via annotation, not
-		// the SQL body, so the body-inference tiers above miss it. Add it from
-		// the live-parsed annotations so an edited materialize script keeps its
-		// output edge (the loop below dedups against existing assets/edges). A
-		// managed scd2 materialize also produces the `<dim>_current` companion
-		// view — add it too (mirrors the deploy path) so a draft consuming only
-		// the view links back to this producer instead of orphaning.
-		if (parsed.materialize) {
-			writeOuts.push({
-				kind: parsed.materialize.targetKind,
-				path: parsed.materialize.targetPath
-			})
-			const currentPath = scd2CurrentTargetPath(parsed.materialize)
-			if (currentPath) {
-				writeOuts.push({
-					kind: parsed.materialize.targetKind,
-					path: currentPath,
-					derivedFrom: parsed.materialize.targetPath
-				})
-			}
-		}
+		// the SQL body, so the body-inference tiers above miss it (pushWriteOut
+		// dedups against existing assets/edges).
+		writeOuts.push(...materializeOuts(parsed))
 		for (const out of writeOuts) {
-			const existing = assets.find((a) => a.kind === out.kind && a.path === out.path)
-			if (!existing) {
-				assets.push({
-					kind: out.kind,
-					path: out.path,
-					...(out.derivedFrom ? { derived_from: out.derivedFrom } : {})
-				})
-			} else if (out.derivedFrom && existing.derived_from == undefined) {
-				existing.derived_from = out.derivedFrom
-			}
-			// Dedup against edges already in the overlay (this draft's base
-			// edges were dropped above, so this only guards against duplicate
-			// writeOuts entries — not against the persisted version).
-			const hasWriteEdge = edges.some(
-				(e) =>
-					e.runnable_kind === 'script' &&
-					e.runnable_path === path &&
-					e.asset_kind === out.kind &&
-					e.asset_path === out.path &&
-					(e.access_type === 'w' || e.access_type === 'rw')
-			)
-			if (hasWriteEdge) continue
-			edges.push({
-				runnable_path: path,
-				runnable_kind: 'script',
-				asset_kind: out.kind,
-				asset_path: out.path,
-				access_type: 'w',
-				unsaved: true
-			})
+			pushWriteOut(acc, path, out)
 		}
 		// Live read lineage for the active draft (body reads like loadS3File /
 		// SELECT). Only the open draft has live-inferred assets; inactive drafts
@@ -549,26 +599,7 @@ function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
 		// edges of a saved script the moment the user starts editing it.
 		if (liveForThisDraft) {
 			for (const inp of extractReads(liveBodyAssets.assets)) {
-				if (!assets.some((a) => a.kind === inp.kind && a.path === inp.path)) {
-					assets.push({ kind: inp.kind, path: inp.path })
-				}
-				const hasReadEdge = edges.some(
-					(e) =>
-						e.runnable_kind === 'script' &&
-						e.runnable_path === path &&
-						e.asset_kind === inp.kind &&
-						e.asset_path === inp.path &&
-						(e.access_type === 'r' || e.access_type === 'rw')
-				)
-				if (hasReadEdge) continue
-				edges.push({
-					runnable_path: path,
-					runnable_kind: 'script',
-					asset_kind: inp.kind,
-					asset_path: inp.path,
-					access_type: 'r',
-					unsaved: true
-				})
+				pushReadIn(acc, path, inp)
 			}
 		}
 		// Seed trigger edges from the draft's template so the graph stays
@@ -700,6 +731,25 @@ function applyLiveBufferOverlay(acc: Accumulator, input: ResolveGraphInput, ctx:
 			}
 		}
 	}
+	// Lineage overlay for an open SAVED script. seedDraftOverlays owns drafts,
+	// but a deployed script's unsaved edits are only promoted to a draft on
+	// pane teardown — until then the buffer's lineage lives here. staleForOpen
+	// already dropped the base edges the buffer no longer references; this adds
+	// the ones it now does. Without it, retargeting `// materialize` (or a body
+	// write/read) shows no output edge until the user clicks away.
+	if (!ctx.draftedPaths.has(livePath)) {
+		for (const out of materializeOuts(liveAnnotations.annotations)) {
+			pushWriteOut(acc, livePath, out)
+		}
+		if (input.liveBodyAssets.scriptPath === livePath) {
+			for (const out of extractWrites(input.liveBodyAssets.assets)) {
+				pushWriteOut(acc, livePath, out)
+			}
+			for (const inp of extractReads(input.liveBodyAssets.assets)) {
+				pushReadIn(acc, livePath, inp)
+			}
+		}
+	}
 	// Native trigger annotations: kinds for which a matching trigger
 	// row was found in the backend response. If the live buffer
 	// declares `// on kafka` and at least one kafka_trigger row points