diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 4e5acaee08..aab963ad40 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/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>({}) 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>({}) 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. diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts index 6d1b50e3ee..d3c824378b 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts +++ b/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([]) + }) +}) diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts index d610bc6059..4acad8fe03 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts +++ b/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 ` annotation declares: the + * target itself, plus — for a managed scd2 materialize — the `_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 ` 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 `_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