fix(pipelines): live materialize/dataset editing — stale graph, phantom drafts, stale Save-all deploys (#9990)

* fix(pipelines): live materialize/dataset edits reflect on the graph; no phantom draft after deploy Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): Save all deploys the open pane's live buffer, not the stale draft snapshot Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): pin deployedFromPane to the shipped content so mid-deploy keystrokes still promote to a draft Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): draft rename ping-pong loop, stale rename deploys, inactive-draft input lineage Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): dedupe inferred-lineage overlay against accumulated edges; first draft teardown still captures reads Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): record an authoritative empty read capture on uncaptured draft entries Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): teardown skip compares lineage too, so access-only overrides still persist Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * style(pipelines): compress persist-back guard comments to the invariant Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-09-06 00:02:13 +00:00 · 2026-07-07 22:00:39 +02:00
parent a6276b5900
commit f7efb646bf
8 changed files with 574 additions and 134 deletions
@@ -1,5 +1,6 @@
 <script lang="ts">
 	import { ScriptService, type Script, type ScriptLang } from '$lib/gen'
+	import { untrack } from 'svelte'
 	import { resource } from 'runed'
 	import { base } from '$lib/base'
 	import Button from '$lib/components/common/button/Button.svelte'
@@ -53,6 +54,11 @@
 		// `selection` when present. Used by the pipeline + menu so a new
 		// pipeline script opens inline instead of navigating to /scripts/add.
 		draftScript?: Script | undefined
+		// The draft entry's captured lineage — the teardown skip compares against
+		// these, since access overrides change lineage without a text edit and an
+		// uncaptured entry (inputAssets undefined) must still emit once.
+		draftOutputAssets?: Array<{ kind: AssetWithAltAccessType['kind']; path: string }>
+		draftInputAssets?: Array<{ kind: AssetWithAltAccessType['kind']; path: string }>
 		// Local-dev preview (`/pipeline_dev`): resolve a selected node to its
 		// working-tree content instead of fetching the deployed script (there is
 		// none — the pipeline is local-only). Returns a read-only `Script` so the
@@ -119,6 +125,9 @@
 			snapshot: {
 				content: string
 				writes: { kind: AssetWithAltAccessType['kind']; path: string }[]
+				// Body-inferred reads, so the parent's draft keeps its input
+				// lineage while inactive (no live inference runs for it).
+				reads: { kind: AssetWithAltAccessType['kind']; path: string }[]
 				// Full edited script — set when the source is a persisted
 				// script (needed to seed a brand-new draft entry).
 				script?: Script
@@ -257,6 +266,8 @@
 	let {
 		selection,
 		draftScript,
+		draftOutputAssets,
+		draftInputAssets,
 		resolveLocalScript,
 		localScriptsVersion,
 		workspace,
@@ -482,6 +493,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
@@ -520,7 +532,41 @@
 		const origAtRegister = isDraftRun ? undefined : scriptRes.current
 		if (!isDraftRun && !origAtRegister) return
 		const captured = script
+		// Untracked — tracked reads here would re-run (and emit) per keystroke.
+		// Snapshotted at registration so the cleanup compares against ITS entry
+		// (at cleanup time the props may already point at the next draft).
+		const contentAtRegister = untrack(() => captured.content ?? '')
+		const writesAtRegister = untrack(() => draftOutputAssets)
+		const readsAtRegister = untrack(() => draftInputAssets)
+		const refsEq = (
+			a: Array<{ kind: string; path: string }>,
+			b: Array<{ kind: string; path: string }>
+		) => a.length === b.length && a.every((x, i) => x.kind === b[i]?.kind && x.path === b[i]?.path)
 		return () => {
+			const writes = (liveBodyAssets ?? [])
+				.filter((a) => {
+					const t = a.access_type ?? a.alt_access_type
+					return t === 'w' || t === 'rw'
+				})
+				.map((a) => ({ kind: a.kind, path: a.path }))
+			const reads = (liveBodyAssets ?? [])
+				.filter((a) => {
+					const t = a.access_type ?? a.alt_access_type
+					return t === 'r' || t === 'rw'
+				})
+				.map((a) => ({ kind: a.kind, path: a.path }))
+			// Draft runs: emit only when content or lineage changed in THIS clone.
+			// An unconditional emit ping-pongs forever when the entry is rewritten
+			// externally while the pane stays mounted (rename rekey, AI edit):
+			// stale-generation emit → entry rewrite → re-clone → emit, pinning the tab.
+			if (
+				isDraftRun &&
+				(captured.content ?? '') === contentAtRegister &&
+				refsEq(writesAtRegister ?? [], writes) &&
+				readsAtRegister != undefined &&
+				refsEq(readsAtRegister, reads)
+			)
+				return
 			if (!isDraftRun) {
 				// Prefer the freshest fetch when it's still this script
 				// (cleanup reads are untracked): after the pane's own Save +
@@ -532,23 +578,31 @@
 				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) => {
-					const t = a.access_type ?? a.alt_access_type
-					return t === 'w' || t === 'rw'
-				})
-				.map((a) => ({ kind: a.kind, path: a.path }))
 			onDraftPersist?.(captured.path, {
 				content: captured.content ?? '',
 				writes,
+				reads,
 				script: isDraftRun ? undefined : (structuredClone($state.snapshot(captured)) as Script)
 			})
 		}
 	})

-	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
@@ -676,10 +730,17 @@
 			} catch {
 				sendUserToast(`Could not parse code, are you sure it is valid?`, true)
 			}
+			// Pin the exact content this deploy ships: the editor stays live during
+			// the network round-trip, so `script.content` can advance past what was
+			// sent — `deployedFromPane` must record the shipped version, not the
+			// buffer at response time, or mid-deploy keystrokes match the guard and
+			// are never promoted to a draft.
+			const contentAtDeploy = script.content ?? ''
 			const newHash = await ScriptService.createScript({
 				workspace,
 				requestBody: {
 					...script,
+					content: contentAtDeploy,
 					language: script.language,
 					description: script.description ?? '',
 					// Brand-new drafts have no prior hash (buildDraft seeds ''
@@ -714,6 +775,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: contentAtDeploy }
 			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.
@@ -5,7 +5,7 @@
 	import { DraftService } from '$lib/gen'
 	import { decodeState, encodeState } from '$lib/utils'
 	import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
-	import { extractWrites } from '$lib/components/assets/lib'
+	import { extractReads, extractWrites } from '$lib/components/assets/lib'
 	import type { PipelineDraft } from './pipelineAiHelpers'
 	import type { ColumnLineageGraph } from './columnLineageGraph'
 	import HideButton from '$lib/components/apps/editor/settingsPanel/HideButton.svelte'
@@ -369,6 +369,11 @@
 			editor.drafts.has(editor.liveBodyAssets.scriptPath)
 				? extractWrites(editor.liveBodyAssets.assets)
 				: undefined
+		const liveReadsSnapshot =
+			editor.liveBodyAssets.scriptPath != undefined &&
+			editor.drafts.has(editor.liveBodyAssets.scriptPath)
+				? extractReads(editor.liveBodyAssets.assets)
+				: undefined
 		const liveWritesPath = editor.liveBodyAssets.scriptPath
 		const liveContentPath =
 			editor.liveContent.scriptPath != undefined && editor.drafts.has(editor.liveContent.scriptPath)
@@ -382,13 +387,15 @@
 						? liveWritesSnapshot
 						: undefined
 					: d.outputAssets
+			const inputAssets =
+				liveReadsSnapshot != undefined && liveWritesPath === p ? liveReadsSnapshot : d.inputAssets
 			const script =
 				liveContentPath === p && d.script.content !== liveContentValue
 					? { ...d.script, content: liveContentValue }
 					: d.script
-			if (script === d.script && outputAssets === d.outputAssets)
+			if (script === d.script && outputAssets === d.outputAssets && inputAssets === d.inputAssets)
 				return [p, d] as [string, PipelineDraft]
-			return [p, { ...d, script, outputAssets }] as [string, PipelineDraft]
+			return [p, { ...d, script, outputAssets, inputAssets }] as [string, PipelineDraft]
 		})
 		const activePath = editor.activeDraftPath
 		const key = storageKey
@@ -524,6 +531,8 @@
 						{requestRunCascadeSignal}
 						{focusUploadSignal}
 						draftScript={activeDraft?.script}
+						draftOutputAssets={activeDraft?.outputAssets}
+						draftInputAssets={activeDraft?.inputAssets}
 						{pathPrefix}
 						{onDraftPathChange}
 						{workspace}
@@ -1,7 +1,11 @@
 import { JobService, ScriptService, type AssetKind, type Script, type ScriptLang } from '$lib/gen'
 import { emptySchema, sendUserToast } from '$lib/utils'
 import { inferAssets } from '$lib/infer'
-import { extractWrites, type AssetWithAltAccessType } from '$lib/components/assets/lib'
+import {
+	extractReads,
+	extractWrites,
+	type AssetWithAltAccessType
+} from '$lib/components/assets/lib'
 import { assetUri, autoOutputAsset, type PipelineOutputKind } from './pipelineTemplates'
 import { parsePipelineAnnotations } from './parsePipelineAnnotations'
 import type { AssetGraphResponse } from './types'
@@ -31,6 +35,11 @@ export type PipelineDraft = {
 	localId: string
 	script: Script
 	outputAssets?: Array<{ kind: AssetKind; path: string }>
+	/** Body-inferred reads captured with the draft, so an inactive draft keeps
+	 * its input lineage on the canvas. `undefined` = not captured yet (legacy
+	 * bundle / just-seeded draft) — consumers fall back to the session cache;
+	 * an empty array is an authoritative "reads nothing". */
+	inputAssets?: Array<{ kind: AssetKind; path: string }>
 }

 export type PipelineAiHelperDeps = {
@@ -81,16 +90,20 @@ export function makePipelineScript(
 	} as unknown as Script
 }

-async function inferOutputAssets(
+async function inferDraftAssets(
 	language: ScriptLang,
 	content: string
-): Promise<Array<{ kind: AssetKind; path: string }>> {
+): Promise<{
+	writes: Array<{ kind: AssetKind; path: string }>
+	reads: Array<{ kind: AssetKind; path: string }>
+}> {
 	try {
 		const inferred = await inferAssets(language, content)
-		if (inferred?.status === 'error') return []
-		return extractWrites((inferred?.assets ?? []) as AssetWithAltAccessType[])
+		if (inferred?.status === 'error') return { writes: [], reads: [] }
+		const assets = (inferred?.assets ?? []) as AssetWithAltAccessType[]
+		return { writes: extractWrites(assets), reads: extractReads(assets) }
 	} catch {
-		return []
+		return { writes: [], reads: [] }
 	}
 }

@@ -219,11 +232,11 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 					)
 				}
 			}
-			const inferred = await inferOutputAssets(language, content)
+			const inferred = await inferDraftAssets(language, content)
 			// Fall back to a seeded output (from the declared output_kind) when the
 			// body doesn't yet write anything inferable.
 			const seeded =
-				inferred[0] ??
+				inferred.writes[0] ??
 				(outputKind
 					? autoOutputAsset(outputKind as PipelineOutputKind, deps.getFolder(), language)
 					: undefined)
@@ -231,7 +244,8 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 			next.set(path, {
 				localId: deps.newDraftLocalId(),
 				script: makePipelineScript(language, path, content, new Date().toISOString()),
-				outputAssets: inferred.length > 0 ? inferred : seeded ? [seeded] : undefined
+				outputAssets: inferred.writes.length > 0 ? inferred.writes : seeded ? [seeded] : undefined,
+				inputAssets: inferred.reads
 			})
 			deps.setDrafts(next)
 			deps.onShowDrafts?.()
@@ -258,12 +272,13 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
 				if (!workspace) throw new Error('No workspace is selected.')
 				baseScript = await ScriptService.getScriptByPath({ workspace, path })
 			}
-			const inferred = await inferOutputAssets(baseScript.language, content)
+			const inferred = await inferDraftAssets(baseScript.language, content)
 			const next = new Map(drafts)
 			next.set(path, {
 				localId: existing?.localId ?? deps.newDraftLocalId(),
 				script: { ...baseScript, content },
-				outputAssets: inferred.length > 0 ? inferred : existing?.outputAssets
+				outputAssets: inferred.writes.length > 0 ? inferred.writes : existing?.outputAssets,
+				inputAssets: inferred.reads
 			})
 			deps.setDrafts(next)
 			deps.onShowDrafts?.()
@@ -132,7 +132,14 @@ export class PipelineEditorState {
 	 * entry). Verbatim port of the route page's `handleDraftPersist`. */
 	handleDraftPersist = (
 		p: string,
-		snapshot: { content: string; writes: { kind: AssetKind; path: string }[]; script?: Script }
+		snapshot: {
+			content: string
+			writes: { kind: AssetKind; path: string }[]
+			// Optional: undefined = reads not captured by this caller — keep
+			// whatever the draft already carries.
+			reads?: { kind: AssetKind; path: string }[]
+			script?: Script
+		}
 	) => {
 		queueMicrotask(() => {
 			const d = this.drafts.get(p)
@@ -142,7 +149,8 @@ export class PipelineEditorState {
 				next.set(p, {
 					localId: this.newDraftLocalId(),
 					script: snapshot.script,
-					outputAssets: snapshot.writes.length > 0 ? snapshot.writes : undefined
+					outputAssets: snapshot.writes.length > 0 ? snapshot.writes : undefined,
+					inputAssets: snapshot.reads
 				})
 				this.drafts = next
 				return
@@ -153,17 +161,26 @@ export class PipelineEditorState {
 			// every persist re-writes the drafts Map with an equivalent object,
 			// re-triggering the pane's emit → graph re-derive → persist, an infinite
 			// microtask loop (hangs the tab without an effect-depth throw).
-			const writesEqual =
-				(d.outputAssets?.length ?? 0) === snapshot.writes.length &&
-				(d.outputAssets ?? []).every(
-					(a, i) => a.kind === snapshot.writes[i]?.kind && a.path === snapshot.writes[i]?.path
-				)
-			if (d.script.content === snapshot.content && writesEqual) return
+			const refsEqual = (
+				a: Array<{ kind: AssetKind; path: string }>,
+				b: Array<{ kind: AssetKind; path: string }>
+			) =>
+				a.length === b.length && a.every((x, i) => x.kind === b[i]?.kind && x.path === b[i]?.path)
+			const writesEqual = refsEqual(d.outputAssets ?? [], snapshot.writes)
+			// An uncaptured entry (`inputAssets` undefined) is never "equal" to an
+			// incoming capture — even `reads: []` must be recorded, or the entry
+			// stays on the legacy fallback (session cache) and can keep stale read
+			// edges after the pane closes.
+			const readsEqual =
+				snapshot.reads == undefined ||
+				(d.inputAssets != undefined && refsEqual(d.inputAssets, snapshot.reads))
+			if (d.script.content === snapshot.content && writesEqual && readsEqual) return
 			const next = new Map(this.drafts)
 			next.set(p, {
 				...d,
 				script: { ...d.script, content: snapshot.content },
-				outputAssets: snapshot.writes.length > 0 ? snapshot.writes : undefined
+				outputAssets: snapshot.writes.length > 0 ? snapshot.writes : undefined,
+				inputAssets: snapshot.reads ?? d.inputAssets
 			})
 			this.drafts = next
 		})
@@ -68,3 +68,39 @@ describe('PipelineEditorState.handleDraftPersist', () => {
 		expect(pe.drafts).toBe(before)
 	})
 })
+
+describe('PipelineEditorState.handleDraftPersist — read capture', () => {
+	it('records an authoritative empty capture on an uncaptured entry (undefined → [])', async () => {
+		const pe = new PipelineEditorState()
+		pe.drafts = new Map([['f/x/n', draft('SELECT 1', undefined)]])
+		const before = pe.drafts
+		pe.handleDraftPersist('f/x/n', { content: 'SELECT 1', writes: [], reads: [] })
+		await flushMicrotasks()
+		// Must rewrite: undefined means "fall back to the session cache", [] means
+		// "reads nothing" — staying undefined would keep stale cached read edges
+		// alive after the pane closes.
+		expect(pe.drafts).not.toBe(before)
+		expect(pe.drafts.get('f/x/n')?.inputAssets).toEqual([])
+	})
+
+	it('is idempotent once the capture matches ([] vs reads: [])', async () => {
+		const pe = new PipelineEditorState()
+		const d = draft('SELECT 1', undefined)
+		d.inputAssets = []
+		pe.drafts = new Map([['f/x/n', d]])
+		const before = pe.drafts
+		pe.handleDraftPersist('f/x/n', { content: 'SELECT 1', writes: [], reads: [] })
+		await flushMicrotasks()
+		expect(pe.drafts).toBe(before)
+	})
+
+	it('is idempotent when a legacy caller omits reads', async () => {
+		const pe = new PipelineEditorState()
+		pe.drafts = new Map([['f/x/n', draft('SELECT 1', undefined)]])
+		const before = pe.drafts
+		pe.handleDraftPersist('f/x/n', { content: 'SELECT 1', writes: [] })
+		await flushMicrotasks()
+		expect(pe.drafts).toBe(before)
+		expect(pe.drafts.get('f/x/n')?.inputAssets).toBeUndefined()
+	})
+})
@@ -796,3 +796,231 @@ 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([])
+	})
+})
+
+describe('inactive draft input lineage', () => {
+	it('keeps read edges + derived cascade for a deselected draft via inputAssets', () => {
+		const drafts = new Map([
+			[
+				'f/x/cons',
+				{
+					script: { content: '-- pipeline\n-- materialize ducklake://main/agg\nselect 1' },
+					outputAssets: [{ kind: 'ducklake' as const, path: 'main/agg' }],
+					inputAssets: [{ kind: 'ducklake' as const, path: 'main/src' }]
+				}
+			]
+		])
+		// No live overlay: the draft is NOT the open script.
+		const r = resolveGraph(input({ drafts }))
+		expect(r.assets).toContainEqual({ kind: 'ducklake', path: 'main/src' })
+		expect(r.edges).toContainEqual({
+			runnable_path: 'f/x/cons',
+			runnable_kind: 'script',
+			asset_kind: 'ducklake',
+			asset_path: 'main/src',
+			access_type: 'r',
+			unsaved: true
+		})
+		// Auto-derived cascade trigger from the captured read.
+		expect(assetTrigKeys(r, 'f/x/cons')).toContain('ducklake:main/src')
+	})
+
+	it('an empty inputAssets array is authoritative — no session-cache fallback', () => {
+		const drafts = new Map([
+			[
+				'f/x/cons',
+				{
+					script: { content: '-- pipeline\nselect 1' },
+					inputAssets: [] as Array<{ kind: 'ducklake'; path: string }>
+				}
+			]
+		])
+		const inferredReadsByPath = new Map([
+			['f/x/cons', [{ kind: 'ducklake' as const, path: 'main/stale' }]]
+		])
+		const r = resolveGraph(input({ drafts, inferredReadsByPath }))
+		expect(r.assets).not.toContainEqual({ kind: 'ducklake', path: 'main/stale' })
+		expect(r.edges.filter((e) => e.asset_path === 'main/stale')).toEqual([])
+	})
+})
+
+describe('open saved script: live overlay vs inferred-lineage maps', () => {
+	it('does not duplicate edges when the same live ref reaches both paths', () => {
+		// The route page mirrors the open pane's liveBodyAssets into
+		// inferredReads/WritesByPath, so the same refs arrive twice.
+		const base = baseGraph({
+			runnables: [{ path: 'f/x/cons', usage_kind: 'script', in_pipeline: true }]
+		})
+		const live = [duck('main/src', 'r'), s3('/out.parquet', 'w')]
+		const r = resolveGraph(
+			input({
+				base,
+				liveBodyAssets: { scriptPath: 'f/x/cons', assets: live },
+				liveAnnotations: { scriptPath: 'f/x/cons', annotations: ann({ inPipeline: true }) },
+				inferredReadsByPath: new Map([
+					['f/x/cons', [{ kind: 'ducklake' as const, path: 'main/src' }]]
+				]),
+				inferredWritesByPath: new Map([
+					['f/x/cons', [{ kind: 's3object' as const, path: '/out.parquet' }]]
+				])
+			})
+		)
+		expect(
+			r.edges.filter((e) => e.asset_path === 'main/src' && e.runnable_path === 'f/x/cons')
+		).toHaveLength(1)
+		expect(
+			r.edges.filter((e) => e.asset_path === '/out.parquet' && e.runnable_path === 'f/x/cons')
+		).toHaveLength(1)
+	})
+})
@@ -17,6 +17,9 @@ import {
 export type GraphDraft = {
 	script: { content: string }
 	outputAssets?: Array<{ kind: AssetKind; path: string }>
+	/** Reads captured on pane teardown. `undefined` = not captured (legacy
+	 * bundle) → fall back to the session cache; `[]` = authoritative none. */
+	inputAssets?: Array<{ kind: AssetKind; path: string }>
 }

 export type ResolveGraphInput = {
@@ -408,6 +411,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 +508,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,86 +589,21 @@ 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
-		// fall back to their `// on <asset>` annotations below for inputs. Without
-		// this, dropping the persisted base edges above would lose the input
-		// 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
-				})
-			}
+		// Read lineage: live inference for the open draft, the captured
+		// `inputAssets` snapshot for inactive ones (with the session cache as a
+		// legacy fallback). Without the inactive tier, merely selecting another
+		// node would drop this draft's input edges from the canvas.
+		const draftReads = liveForThisDraft
+			? extractReads(liveBodyAssets.assets)
+			: (d.inputAssets ?? inferredReadsByPath.get(path) ?? [])
+		for (const inp of draftReads) {
+			pushReadIn(acc, path, inp)
 		}
 		// Seed trigger edges from the draft's template so the graph stays
 		// stable when the user clicks off this draft. Live annotations
@@ -592,17 +626,14 @@ function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
 			if (!hasTriggerAsset) assets.push({ kind: a.kind, path: a.path })
 		}
 		// Auto-derived cascade edges (backend parity): a ducklake/s3 read wires
-		// the edge from the body alone. Reads/writes come from the active draft's
-		// live inference, else the sticky session cache. The open buffer's derived
-		// edges are re-computed authoritatively in applyLiveBufferOverlay (which
-		// strips this path's seeded triggers first), same as the explicit `// on`
-		// triggers above.
-		const draftReads = liveForThisDraft
-			? extractReads(liveBodyAssets.assets)
-			: (inferredReadsByPath.get(path) ?? [])
+		// the edge from the body alone. Reads reuse the tier above; writes come
+		// from the active draft's live inference, else the captured snapshot /
+		// session cache. The open buffer's derived edges are re-computed
+		// authoritatively in applyLiveBufferOverlay (which strips this path's
+		// seeded triggers first), same as the explicit `// on` triggers above.
 		const draftWrites = liveForThisDraft
 			? extractWrites(liveBodyAssets.assets)
-			: (inferredWritesByPath.get(path) ?? [])
+			: (d.outputAssets ?? inferredWritesByPath.get(path) ?? [])
 		for (const a of deriveAutoAssetTriggers(draftReads, draftWrites, parsed)) {
 			extraTriggers.push({
 				trigger_kind: 'asset',
@@ -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
@@ -748,7 +798,7 @@ function crossCheckSweptScripts(acc: Accumulator, input: ResolveGraphInput) {
 * changes — not just while selected.
 */
 function overlayInferredLineage(acc: Accumulator, input: ResolveGraphInput) {
-	const { base, drafts, inferredWritesByPath, inferredReadsByPath } = input
+	const { drafts, inferredWritesByPath, inferredReadsByPath } = input
 	const { assets, edges } = acc

 	const overlayLineage = (
@@ -757,19 +807,21 @@ function overlayInferredLineage(acc: Accumulator, input: ResolveGraphInput) {
 	) => {
 		for (const [scriptPath, refs] of byPath) {
 			if (drafts.has(scriptPath)) continue
-			const persisted = new Set(
-				base.edges
-					.filter(
-						(e) =>
-							e.runnable_path === scriptPath &&
-							e.runnable_kind === 'script' &&
-							(e.access_type === access || e.access_type === 'rw')
-					)
-					.map((e) => `${e.asset_kind}:${e.asset_path}`)
-			)
 			for (const a of refs) {
-				const key = `${a.kind}:${a.path}`
-				if (persisted.has(key)) continue
+				// Dedup against the ACCUMULATED edges, not just base: for the open
+				// script these same refs may already be overlaid by
+				// applyLiveBufferOverlay (which feeds the maps on the route page),
+				// and a duplicate would collide on the canvas's endpoint-derived
+				// edge ids.
+				const hasEdge = edges.some(
+					(e) =>
+						e.runnable_path === scriptPath &&
+						e.runnable_kind === 'script' &&
+						e.asset_kind === a.kind &&
+						e.asset_path === a.path &&
+						(e.access_type === access || e.access_type === 'rw')
+				)
+				if (hasEdge) continue
 				if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
 					assets.push({ kind: a.kind, path: a.path })
 				}
@@ -654,7 +654,22 @@
 		if (!$workspaceStore || pe.drafts.size === 0 || savingAll) return
 		savingAll = true
 		const ws = $workspaceStore
-		const entries = [...pe.drafts.entries()]
+		// The open pane's keystrokes live in `pe.liveContent` until the pane is
+		// torn down — the drafts Map still holds the pre-edit snapshot. Deploy
+		// what the user sees: fold the live buffer into its draft (same merge
+		// the autosave bundle applies before persisting).
+		const liveContentPath =
+			pe.liveContent.scriptPath != undefined && pe.drafts.has(pe.liveContent.scriptPath)
+				? pe.liveContent.scriptPath
+				: undefined
+		const entries = [...pe.drafts.entries()].map(([path, d]) => {
+			if (path !== liveContentPath || d.script.content === pe.liveContent.content)
+				return [path, d] as [string, Draft]
+			return [path, { ...d, script: { ...d.script, content: pe.liveContent.content } }] as [
+				string,
+				Draft
+			]
+		})
 		// Snapshot what the preview promises for every draft before anything
 		// deploys — used to verify the persisted graph below.
 		const predicted = predictCascadeFacts(entries.map(([p]) => p))
@@ -771,7 +786,13 @@
 		// position so the canvas / lists don't reshuffle on rename.
 		for (const [k, v] of pe.drafts) {
 			if (k === oldPath) {
-				const updatedScript = { ...v.script, path: newPath }
+				// Fold the open pane's live buffer in: the drafts Map only syncs
+				// on pane teardown, and the rename both re-clones the pane from
+				// this entry AND auto-deploys it — a stale snapshot here wipes
+				// the user's unsaved keystrokes and ships pre-edit content.
+				const content =
+					pe.liveContent.scriptPath === oldPath ? pe.liveContent.content : v.script.content
+				const updatedScript = { ...v.script, path: newPath, content }
 				next.set(newPath, { ...v, script: updatedScript })
 			} else {
 				next.set(k, v)