From dc7bdd9fac371e7796a50bc003f9ba7056c8f470 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Tue, 7 Jul 2026 18:06:03 +0000
Subject: [PATCH] fix(pipelines): draft rename ping-pong loop, stale rename
 deploys, inactive-draft input lineage

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../AssetGraph/AssetGraphDetailsPane.svelte   | 23 ++++++++++
 .../AssetGraph/PipelineGraphEditor.svelte     | 13 ++++--
 .../assets/AssetGraph/pipelineAiHelpers.ts    | 37 ++++++++++-----
 .../AssetGraph/pipelineEditorState.svelte.ts  | 30 ++++++++----
 .../assets/AssetGraph/resolveGraph.test.ts    | 46 +++++++++++++++++++
 .../assets/AssetGraph/resolveGraph.ts         | 36 +++++++--------
 .../(logged)/pipeline/[folder]/+page.svelte   |  8 +++-
 7 files changed, 151 insertions(+), 42 deletions(-)

diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
index 87ad9aa371..d4dc1a8943 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
@@ -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'
@@ -119,6 +120,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
@@ -521,7 +525,19 @@
 		const origAtRegister = isDraftRun ? undefined : scriptRes.current
 		if (!isDraftRun && !origAtRegister) return
 		const captured = script
+		// Untracked — reading `.content` in the effect body would re-run (and
+		// emit) on every keystroke, the exact per-key loop the header comment
+		// forbids.
+		const contentAtRegister = untrack(() => captured.content ?? '')
 		return () => {
+			// Draft runs: only emit when the user actually typed into THIS clone.
+			// When the drafts entry is rewritten externally while the pane stays
+			// mounted (rename rekey, AI edit), the pane re-clones and this cleanup
+			// fires with a captured generation the user never touched — emitting
+			// it would push the PREVIOUS generation's content back into the entry,
+			// which re-clones and re-emits forever (a microtask ping-pong that
+			// pins the tab and spams the draft autosave endpoint).
+			if (isDraftRun && (captured.content ?? '') === contentAtRegister) return
 			if (!isDraftRun) {
 				// Prefer the freshest fetch when it's still this script
 				// (cleanup reads are untracked): after the pane's own Save +
@@ -550,9 +566,16 @@
 					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 }))
 			onDraftPersist?.(captured.path, {
 				content: captured.content ?? '',
 				writes,
+				reads,
 				script: isDraftRun ? undefined : (structuredClone($state.snapshot(captured)) as Script)
 			})
 		}
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
index cc44d0b41a..37b79852e5 100644
--- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
@@ -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
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts
index 959992543f..3658bc5a32 100644
--- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts
@@ -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?.()
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineEditorState.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineEditorState.svelte.ts
index bdecadff91..ffb2e49733 100644
--- a/frontend/src/lib/components/assets/AssetGraph/pipelineEditorState.svelte.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/pipelineEditorState.svelte.ts
@@ -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,21 @@ 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)
+			const readsEqual =
+				snapshot.reads == 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
 		})
diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts
index d3c824378b..5629da692b 100644
--- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts
@@ -948,3 +948,49 @@ describe('live buffer overlays (open script)', () => {
 		).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([])
+	})
+})
diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts
index 4acad8fe03..196d4f89e6 100644
--- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts
@@ -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 = {
@@ -592,15 +595,15 @@ function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
 		for (const out of writeOuts) {
 			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)) {
-				pushReadIn(acc, path, inp)
-			}
+		// 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
@@ -623,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',
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
index fc75d1ed9f..0c1a6eeb8c 100644
--- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
@@ -786,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)