fix(pipelines): draft rename ping-pong loop, stale rename deploys, inactive-draft input lineage

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-07-07 18:06:03 +00:00
parent 67421f9ec3
commit dc7bdd9fac
7 changed files with 151 additions and 42 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'
@@ -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)
})
}
@@ -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
@@ -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,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
})
@@ -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([])
})
})
@@ -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',
@@ -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)