feat: deploy and discard a session's pipeline edits as one changes-list row

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-30 20:39:57 +02:00
co-authored by Claude Opus 5.5
parent 76d1bd2465
commit dedffeb046
16 changed files with 616 additions and 253 deletions
@@ -0,0 +1,49 @@
<script lang="ts">
import { AlertTriangle } from 'lucide-svelte'
import Popover from '$lib/components/meltComponents/Popover.svelte'
// Which drafts failed to deploy and why, anchored next to the deploy button so
// the reasons are one click away without leaving the editor. Deployed drafts
// leave the map; what is listed is still unresolved.
let {
errors,
open = $bindable()
}: {
errors: ReadonlyMap<string, string>
open?: boolean
} = $props()
</script>
<Popover
placement="bottom-end"
contentClasses="p-3 max-w-[480px]"
usePointerDownOutside
enableFlyTransition
bind:isOpen={open}
>
{#snippet trigger()}
<button
type="button"
class="flex items-center gap-1.5 px-2 py-1 rounded-md text-red-600 dark:text-red-400 bg-red-50 dark:bg-red-900/30 hover:bg-red-100 dark:hover:bg-red-900/50 transition-colors text-xs font-medium"
title="View save errors"
>
<AlertTriangle size={14} />
<span>{errors.size} failed</span>
</button>
{/snippet}
{#snippet content()}
<div class="flex flex-col gap-2">
<span class="text-xs font-semibold text-emphasis">Save errors</span>
<div class="flex flex-col gap-2 max-h-72 overflow-y-auto">
{#each [...errors.entries()] as [path, message] (path)}
<div class="flex flex-col gap-0.5 border-l-2 border-red-400 pl-2">
<span class="text-2xs font-mono text-emphasis">{path}</span>
<span class="text-2xs text-red-600 dark:text-red-400 break-words">
{message}
</span>
</div>
{/each}
</div>
</div>
{/snippet}
</Popover>
@@ -0,0 +1,47 @@
<script lang="ts">
import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
import type { PipelineTriggerDraft } from './types'
// Triggers start running once deployed, so deploying a pipeline's trigger
// drafts asks first, listing what will start.
let {
open,
triggerDrafts,
onConfirmed,
onCanceled
}: {
open: boolean
triggerDrafts: ReadonlyMap<string, PipelineTriggerDraft>
onConfirmed: () => void
onCanceled: () => void
} = $props()
</script>
<ConfirmationModal
{open}
title="Deploy triggers?"
confirmationText="Deploy"
type="info"
{onConfirmed}
{onCanceled}
>
<div class="flex flex-col gap-2 text-xs text-secondary">
<p>
{triggerDrafts.size === 1
? 'Saving the pipeline also creates this trigger. It starts running its script once deployed.'
: `Saving the pipeline also creates these ${triggerDrafts.size} triggers. They start running their scripts once deployed.`}
</p>
<ul class="flex flex-col gap-1">
{#each [...triggerDrafts] as [key, d] (key)}
<li class="flex flex-col">
<span class="font-mono text-2xs text-emphasis truncate">{d.config.path}</span>
<span class="text-2xs text-hint truncate">
{d.kind}{#if d.kind === 'schedule'}
· <span class="font-mono">{d.config.schedule}</span> ({d.config.timezone}){/if}
→ {d.config.script_path}
</span>
</li>
{/each}
</ul>
</div>
</ConfirmationModal>
@@ -1,6 +1,10 @@
<script lang="ts">
import { untrack, type Snippet } from 'svelte'
import { PIPELINE_DRAFT_KIND, pipelineBundlePath } from '$lib/pipelinePaths'
import {
PIPELINE_DRAFT_KIND,
pipelineBundlePath,
pipelineLocalMirrorKey
} from '$lib/pipelinePaths'
import { Loader2 } from 'lucide-svelte'
import { Pane, Splitpanes } from 'svelte-splitpanes'
import { DraftService } from '$lib/gen'
@@ -365,7 +369,7 @@
// for the one-time migration below; the DB is the source of truth on load.
// FlowBuilder's autosave analogue — gated by `persistDrafts`.
let pipelineDraftPath = $derived(pipelineBundlePath(folder))
let storageKey = $derived(`pipeline-${folder}`)
let storageKey = $derived(pipelineLocalMirrorKey(folder))
type PipelineDraftBundle = {
drafts: Array<[string, PipelineDraft]>
activeDraftPath?: string
@@ -0,0 +1,205 @@
import { DraftService, ScriptService, type Script } from '$lib/gen'
import { PIPELINE_DRAFT_KIND, pipelineBundlePath } from '$lib/pipelinePaths'
import type { PipelineTriggerDraft } from './types'
import type { Schema } from '$lib/common'
import { emptySchema } from '$lib/utils'
import { inferArgs, inferAssets } from '$lib/infer'
import type { AssetWithAltAccessType } from '$lib/components/assets/lib'
import type { PipelineDraft } from './pipelineAiHelpers'
import type { PipelineEditorState } from './pipelineEditorState.svelte'
import { deployTriggerDraft } from './pipelineTriggerDraftDeploy'
// Deploying a pipeline's drafts: the scripts first, then the trigger drafts whose
// script is now deployed. Shared by the pipeline page's "Save all" and the AI
// session's "Deploy pipeline", so both deploy exactly the same way.
/** Deploy one pipeline script draft. */
export async function deployPipelineScript(draft: PipelineDraft, workspace: string): Promise<void> {
const script = structuredClone($state.snapshot(draft.script) as Script)
script.schema = script.schema ?? emptySchema()
try {
const result = await inferArgs(script.language, script.content, script.schema as Schema)
;(script as any).auto_kind = result?.auto_kind || undefined
script.has_preprocessor = result?.has_preprocessor ?? false
} catch {
// Inference failures don't block deploys (the same fallback the per-pane
// save uses): createScript is the real validation gate.
}
// Re-infer the lineage from the CURRENT body. `script.assets` is a snapshot
// that isn't refreshed on edit, so a renamed/removed output would otherwise be
// re-deployed as a phantom write edge.
let assets: AssetWithAltAccessType[] = []
try {
const inferred = await inferAssets(script.language, script.content)
if (inferred?.status !== 'error') assets = (inferred?.assets ?? []) as AssetWithAltAccessType[]
} catch {
// An unparsable body deploys with no lineage rather than the stale snapshot.
}
await ScriptService.createScript({
workspace,
requestBody: {
...script,
language: script.language,
description: script.description ?? '',
// A brand-new draft has an empty hash (the backend rejects an empty hex
// string); a draft of a deployed script chains off its hash, since
// createScript rejects a parentless deploy to an occupied path.
parent_hash: script.hash ? String(script.hash) : undefined,
is_template: false,
tag: script.tag,
kind: script.kind as Script['kind'] | undefined,
lock: undefined,
assets: assets as any
}
})
}
// Every field a deploy sends: a draft differing from the live script in any of
// them still has something to deploy.
const DEPLOYED_FIELDS = [
'content',
'language',
'summary',
'description',
'tag',
'kind',
'labels'
] as const
/** Whether a draft equals the live script at its path, i.e. has nothing to deploy. */
export async function matchesDeployedScript(
path: string,
d: PipelineDraft,
workspace: string
): Promise<boolean> {
try {
const live = await ScriptService.getScriptByPath({ workspace, path })
const norm = (v: unknown) =>
JSON.stringify(Array.isArray(v) && v.length === 0 ? null : (v ?? null))
return DEPLOYED_FIELDS.every(
(f) =>
norm(f === 'description' ? (live[f] ?? '') : live[f]) ===
norm(f === 'description' ? (d.script[f] ?? '') : d.script[f])
)
} catch {
return false
}
}
/**
* A folder's draft bundle against what is deployed, one entry per node (and the
* trigger drafts), for a diff viewer: the unit a session reviews and deploys.
*/
export async function loadPipelineDraftDiff(
workspace: string,
folder: string
): Promise<{ before: unknown; after: unknown }> {
const row = await DraftService.getOwnDraft({
workspace,
kind: PIPELINE_DRAFT_KIND,
path: pipelineBundlePath(folder)
}).catch(() => undefined)
const bundle = (row?.value ?? {}) as {
drafts?: Array<[string, PipelineDraft]>
triggerDrafts?: PipelineTriggerDraft[]
}
const drafts = Array.isArray(bundle.drafts) ? bundle.drafts : []
const before: Record<string, string> = {}
const after: Record<string, string> = {}
await Promise.all(
drafts.map(async ([path, d]) => {
after[path] = d.script?.content ?? ''
const live = await ScriptService.getScriptByPath({ workspace, path }).catch(() => undefined)
if (live) before[path] = live.content
})
)
const triggers = (bundle.triggerDrafts ?? []).map((t) => ({
kind: t.kind,
path: t.config.path,
script_path: t.config.script_path
}))
return {
before: { nodes: before },
after: { nodes: after, ...(triggers.length > 0 ? { new_triggers: triggers } : {}) }
}
}
export type PipelineDeployOutcome = {
savedPaths: string[]
/** Keys of the trigger drafts that deployed. */
savedTriggers: string[]
/** Failures by script or trigger path. */
errors: Map<string, string>
}
/**
* Deploy every draft of `editor` into `workspace`, dropping from it what deployed:
* failed drafts stay so they can be fixed and retried. The open pane's unsaved
* keystrokes are deployed with their draft.
*/
export async function deployPipelineDrafts(
editor: PipelineEditorState,
workspace: string
): Promise<PipelineDeployOutcome> {
const live = editor.liveContent
const liveContentPath =
live.scriptPath != undefined && editor.drafts.has(live.scriptPath) ? live.scriptPath : undefined
const entries = [...editor.drafts.entries()].map(([path, d]): [string, PipelineDraft] =>
path !== liveContentPath || d.script.content === live.content
? [path, d]
: [path, { ...d, script: { ...d.script, content: live.content } }]
)
const errors = new Map<string, string>()
const savedPaths: string[] = []
// Parallel: every createScript is independent, and one bad body must not block
// the others.
const results = await Promise.allSettled(
entries.map(([, d]) => deployPipelineScript(d, workspace))
)
for (let i = 0; i < results.length; i++) {
const r = results[i]
const [path, d] = entries[i]
if (r.status === 'fulfilled') {
savedPaths.push(path)
continue
}
// A refused draft that equals the deployed script has nothing left to deploy.
// A duplicate refusal alone does not prove it: it is checked against every
// past version, not just the live one.
if (await matchesDeployedScript(path, d, workspace)) {
savedPaths.push(path)
continue
}
const msg = String((r.reason as any)?.body ?? (r.reason as any)?.message ?? r.reason)
errors.set(
path,
/same hash/i.test(msg)
? 'This is the content of an earlier version, which cannot be deployed again as is. Change anything in it (a comment will do) to deploy it.'
: msg
)
}
// A trigger's `script_path` must already exist, so triggers whose script is
// still an undeployed draft stay drafts.
const ready = [...editor.triggerDrafts].filter(
([, d]) => !editor.drafts.has(d.config.script_path) || savedPaths.includes(d.config.script_path)
)
const triggerResults = await Promise.all(ready.map(([, d]) => deployTriggerDraft(d, workspace)))
const savedTriggers: string[] = []
triggerResults.forEach((ok, i) => {
const [key, d] = ready[i]
if (ok) savedTriggers.push(key)
else errors.set(d.config.path, `Could not create the ${d.kind} trigger, see the notification`)
})
for (const key of savedTriggers) editor.discardTriggerDraft(key)
if (savedPaths.length > 0) {
editor.drafts = new Map([...editor.drafts].filter(([k]) => !savedPaths.includes(k)))
// Keep the pane on the script the user was editing, now deployed.
if (editor.activeDraftPath && savedPaths.includes(editor.activeDraftPath)) {
editor.selection = { kind: 'runnable', runnable_kind: 'script', path: editor.activeDraftPath }
editor.activeDraftPath = undefined
}
}
return { savedPaths, savedTriggers, errors }
}
@@ -97,6 +97,7 @@ describe('pipeline tools', () => {
it('build_pipeline_node forwards to proposeNode and does not deploy', async () => {
const { helpers, calls } = makeHelpers()
const modified: string[] = []
const out = await toolByName('build_pipeline_node').fn({
args: {
path: 'f/analytics/clean',
@@ -106,9 +107,14 @@ describe('pipeline tools', () => {
},
workspace: 'w',
helpers,
toolCallbacks: noopCallbacks(),
toolCallbacks: {
...noopCallbacks(),
onItemModified: (kind, path) => modified.push(`${kind}:${path}`)
},
toolId: 't'
})
// The session tracks the folder's draft bundle, the unit it deploys.
expect(modified).toEqual(['data_pipeline:f/analytics/data_pipeline'])
expect(calls.proposeNode?.[0]?.[0]).toMatchObject({
path: 'f/analytics/clean',
language: 'bun',
@@ -1,7 +1,8 @@
import { z } from 'zod'
import { $ScriptLang } from '$lib/gen/schemas.gen'
import type { ScriptLang } from '$lib/gen'
import { createToolDef, executeTestRun, findAndReplace } from '../shared'
import { createToolDef, executeTestRun, findAndReplace, type ToolCallbacks } from '../shared'
import { PIPELINE_DRAFT_KIND, pipelineBundlePath } from '$lib/pipelinePaths'
import {
NONE,
RUN_PREVIEW,
@@ -250,6 +251,12 @@ const testPipelineNodeToolDef = createToolDef(
{ strict: false }
)
// A pipeline's drafts are one bundle for the whole folder, so that bundle is the
// item a node edit marks as modified — it is what the session deploys as a unit.
function recordPipelineModified(toolCallbacks: ToolCallbacks, pipeline: PipelineAIChatHelpers) {
toolCallbacks.onItemModified?.(PIPELINE_DRAFT_KIND, pipelineBundlePath(pipeline.getFolder()))
}
// Summarize the asset lineage the parser inferred from a just-applied node so the
// model gets same-turn feedback on whether its intended edges formed. An empty
// result is the useful signal: a write/read expressed via a variable or dynamic
@@ -314,6 +321,7 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
content,
outputKind: output_kind as PipelineOutputKind | undefined
})
recordPipelineModified(toolCallbacks, pipeline)
toolCallbacks.setToolStatus(toolId, {
content: `Added draft node '${path}'`,
result: 'Success'
@@ -343,6 +351,7 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
'node source'
)
const { detectedReads, detectedWrites } = await pipeline.editNode(path, updated)
recordPipelineModified(toolCallbacks, pipeline)
toolCallbacks.setToolStatus(toolId, {
content: `Edited draft '${path}'`,
result: 'Success'
@@ -358,6 +367,7 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
const pipeline = pipelineForPath(helpers, path)
toolCallbacks.setToolStatus(toolId, { content: `Discarding draft '${path}'...` })
await pipeline.removeProposedNode(path)
recordPipelineModified(toolCallbacks, pipeline)
toolCallbacks.setToolStatus(toolId, {
content: `Discarded draft '${path}'`,
result: 'Success'
@@ -1,7 +1,7 @@
<script lang="ts">
import { resource } from 'runed'
import { tick, untrack } from 'svelte'
import { Loader2, Workflow } from 'lucide-svelte'
import { Loader2, Save, Workflow } from 'lucide-svelte'
import PipelineGraphEditor from '$lib/components/assets/AssetGraph/PipelineGraphEditor.svelte'
import PipelineTriggerEditors from '$lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte'
import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph'
@@ -14,6 +14,19 @@
import { AssetService, JobService, type AssetKind } from '$lib/gen'
import { sendUserToast } from '$lib/utils'
import { createPipelineAiHelpers } from '$lib/components/assets/AssetGraph/pipelineAiHelpers'
import { deployPipelineDrafts } from '$lib/components/assets/AssetGraph/pipelineDeploy.svelte'
import {
diffDeployedGraph,
extractCascadeFacts,
formatDrift
} from '$lib/components/assets/AssetGraph/deployGraphDiff'
import PipelineDeployErrors from '$lib/components/assets/AssetGraph/PipelineDeployErrors.svelte'
import PipelineDeployTriggersModal from '$lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte'
import AutosaveIndicator from '$lib/components/AutosaveIndicator.svelte'
import { Button } from '$lib/components/common'
import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
import { PIPELINE_DRAFT_KIND, pipelineBundlePath } from '$lib/pipelinePaths'
import type { DeployResult } from '$lib/utils_workspace_deploy'
import type { SessionRuntime } from './sessionRuntime.svelte'
let {
@@ -263,6 +276,88 @@
// as the view is mounted — background tabs included, since each folder's tools
// act on its own state — and release them on unmount.
$effect(() => aiChatManager.setPipelineHelpers(helpers))
// ── Deploy all ───────────────────────────────────────────────────────────
// The same deploy as the pipeline page's "Save all", reached from this tab's
// button and from the session's changes list (through the runtime).
const pendingCount = $derived(pe.drafts.size + pe.triggerDrafts.size)
let deploying = $state(false)
let deployErrors = $state<Map<string, string>>(new Map())
let deployErrorsOpen = $state(false)
let triggerConfirmOpen = $state(false)
let resolveTriggerConfirm: ((ok: boolean) => void) | undefined
function confirmTriggers(): Promise<boolean> {
triggerConfirmOpen = true
return new Promise((resolve) => (resolveTriggerConfirm = resolve))
}
function answerTriggerConfirm(ok: boolean) {
triggerConfirmOpen = false
resolveTriggerConfirm?.(ok)
resolveTriggerConfirm = undefined
}
// A deploy asked for from the changes list can arrive while the tab is still
// loading this folder's drafts.
async function hydrated(timeoutMs = 10000): Promise<boolean> {
for (let waited = 0; !pe.hydratedFromDb && waited < timeoutMs; waited += 100) {
await new Promise((resolve) => setTimeout(resolve, 100))
}
return pe.hydratedFromDb
}
async function deployAll(): Promise<DeployResult> {
if (deploying) return { success: false, error: 'This pipeline is already deploying.' }
if (!(await hydrated())) return { success: false, error: 'The pipeline drafts did not load.' }
if (pendingCount === 0) return { success: true }
if (pe.triggerDrafts.size > 0 && !(await confirmTriggers())) {
return { success: false, error: 'Deploy cancelled.' }
}
deploying = true
try {
// What the canvas promises for each draft, checked against what the
// backend derived once deployed.
const predicted = new Map(
[...pe.drafts.keys()].map((p) => [p, extractCascadeFacts(resolvedGraph, p)])
)
const { savedPaths, savedTriggers, errors } = await deployPipelineDrafts(pe, workspaceId)
if (savedPaths.length > 0 || savedTriggers.length > 0) {
await graphRes.refetch()
const deployed = new Map([...predicted].filter(([p]) => savedPaths.includes(p)))
const drift = formatDrift(
diffDeployedGraph(deployed, (graphRes.current ?? EMPTY_GRAPH) as AssetGraphResponse)
)
if (drift) sendUserToast(drift, true)
}
// The session's changes list re-reads the draft once this resolves, so
// what is left of it has to be saved by then.
await tick()
await UserDraftDbSyncer.flush({
workspace: workspaceId,
itemKind: PIPELINE_DRAFT_KIND,
path: pipelineBundlePath(path)
})
deployErrors = errors
if (errors.size > 0) {
deployErrorsOpen = true
return {
success: false,
error: `${errors.size} of the pipeline's drafts failed to deploy, see its tab.`
}
}
return { success: true }
} finally {
deploying = false
}
}
async function deployAllFromButton() {
const count = pendingCount
const res = await deployAll()
if (res.success) sendUserToast(`Deployed ${count} draft${count === 1 ? '' : 's'}`)
}
$effect(() => runtime.registerPipelineView(path, { deployAll }))
</script>
<div class="flex flex-col h-full w-full bg-surface">
@@ -272,6 +367,31 @@
<Workflow size={14} />
<span class="font-mono text-emphasis truncate">f/{path}</span>
<span class="text-tertiary">· data pipeline</span>
{#if workspaceId}
<AutosaveIndicator
workspace={workspaceId}
itemKind={PIPELINE_DRAFT_KIND}
path={pipelineBundlePath(path)}
draftOnly
loadedFromDraft={pe.loadedFromDbDraft}
/>
{/if}
<div class="flex-1"></div>
{#if deployErrors.size > 0}
<PipelineDeployErrors errors={deployErrors} bind:open={deployErrorsOpen} />
{/if}
{#if pendingCount > 0}
<Button
variant="accent"
unifiedSize="xs"
startIcon={{ icon: deploying ? Loader2 : Save }}
onclick={deployAllFromButton}
disabled={deploying}
title={deploying ? 'Deploying drafts…' : `Deploy all ${pendingCount} drafts`}
>
{deploying ? 'Deploying…' : `Deploy all (${pendingCount})`}
</Button>
{/if}
</div>
<div class="flex-1 min-h-0">
<!-- Only block on the deployed-graph fetch when there's nothing to show yet.
@@ -375,3 +495,10 @@
workspace={workspaceId}
onUpdate={() => graphRes.refetch()}
/>
<PipelineDeployTriggersModal
open={triggerConfirmOpen}
triggerDrafts={pe.triggerDrafts}
onConfirmed={() => answerTriggerConfirm(true)}
onCanceled={() => answerTriggerConfirm(false)}
/>
@@ -22,7 +22,7 @@
import { badgeCounts, badgeOf, buildDeployItems, type DeployItem } from './sessionDeployModel'
import { useExistingMaskKeys } from './sessionDeployModel.svelte'
import { previewTargetForDeployKind } from './sessionPreviewTabs.svelte'
import { pipelineFolderFromBundlePath } from '$lib/pipelinePaths'
import { PIPELINE_DRAFT_KIND, pipelineFolderFromBundlePath } from '$lib/pipelinePaths'
import type { PreviewTarget } from './previewRouter'
import JobsSegment from '$lib/components/copilot/chat/JobsSegment.svelte'
import RowIcon from '$lib/components/common/table/RowIcon.svelte'
@@ -376,6 +376,11 @@
onDataChanged={refreshDock}
onItemDeployed={(item) =>
void runtime?.manager.renameModifiedItem(item.draftKind, item.path, item.displayPath)}
onItemDiscarded={(item) => void runtime?.manager.removeModifiedItem(item.draftKind, item.path)}
onItemDiscarded={(item) => {
void runtime?.manager.removeModifiedItem(item.draftKind, item.path)
const folder = pipelineFolderFromBundlePath(item.path)
if (item.draftKind === PIPELINE_DRAFT_KIND && folder) runtime?.forgetPipelineDrafts(folder)
}}
deployPipeline={runtime?.deployPipeline}
/>
{/if}
@@ -6,6 +6,7 @@
import { devBadgeText } from '$lib/utils/devWorkspaceLabel'
import { useSessionDeployModel } from './sessionDeployModel.svelte'
import type { DeployItem } from './sessionDeployModel'
import type { DeployResult } from '$lib/utils_workspace_deploy'
// Session "Edits" drawer. Builds the deploy model over the session
// workspace's drafts and hands it to WorkspaceDiffDrawer, which renders the
@@ -19,7 +20,8 @@
keys,
onDataChanged,
onItemDeployed,
onItemDiscarded
onItemDiscarded,
deployPipeline
}: {
workspaceId: string
parentWorkspaceId?: string
@@ -36,6 +38,8 @@
* drops it. */
onItemDeployed?: (item: DeployItem) => void
onItemDiscarded?: (item: DeployItem) => void
/** Forwarded to the deploy model: deploys a pipeline folder's draft bundle. */
deployPipeline?: (folder: string) => Promise<DeployResult>
} = $props()
const isFork = $derived(!!parentWorkspaceId)
@@ -51,7 +55,8 @@
mask: keys,
onDataChanged,
onItemDeployed,
onItemDiscarded
onItemDiscarded,
deployPipeline
}))
// Editor URL for a row (every item lives in the session workspace).
@@ -62,6 +62,13 @@ describe('diffInMask', () => {
expect(diffInMask(diff('script', 'u/me/s'), mask)).toBe(true)
})
it('matches the scripts and triggers of a masked pipeline folder', () => {
const mask = new Set(['data_pipeline:f/crm/data_pipeline'])
expect(diffInMask(diff('script', 'f/crm/clean'), mask)).toBe(true)
expect(diffInMask(diff('schedule', 'f/crm/clean_schedule'), mask)).toBe(true)
expect(diffInMask(diff('script', 'f/sales/clean'), mask)).toBe(false)
})
it('matches a legacy app diff under both its identity and bridged mask keys', () => {
expect(diffInMask(diff('app', 'u/me/legacy'), new Set(['app:u/me/legacy']))).toBe(true)
expect(diffInMask(diff('app', 'u/me/legacy'), new Set(['raw_app:u/me/legacy']))).toBe(true)
@@ -1,4 +1,5 @@
import type { UserDraftItemKind, WorkspaceItemDiff } from '$lib/gen'
import { PIPELINE_DRAFT_KIND, pipelineBundlePath } from '$lib/pipelinePaths'
// The "modified items mask" tracks which workspace items an AI chat touched via
// tool calls. Each key is `${UserDraftItemKind}:${storagePath}`. UserDraftItemKind
@@ -49,6 +50,15 @@ export function forkDiffKindToUserDraftKind(kind: ForkDiffKind): UserDraftItemKi
export function diffInMask(diff: WorkspaceItemDiff, mask: Set<string>): boolean {
const kind = forkDiffKindToUserDraftKind(diff.kind)
if (kind !== undefined && mask.has(maskKey(kind, diff.path))) return true
// A pipeline is masked as its folder's draft bundle, which deploys into the
// scripts and triggers of that folder — none of which the mask names itself.
const folder = diff.path.match(/^f\/([^/]+)\//)?.[1]
if (
folder &&
diff.kind !== 'folder' &&
mask.has(maskKey(PIPELINE_DRAFT_KIND, pipelineBundlePath(folder)))
)
return true
// Legacy drag-and-drop apps tally fork diffs under `app`, and an explicit
// `?items=` mask names them `app:<path>` (the same kind the drafts list uses).
// The bridged lookup above reads them as `raw_app` (kept for chat masks, which
@@ -27,6 +27,8 @@ import {
import { maskKey } from './modifiedItemsMask'
import { sessionState } from './sessionState.svelte'
import { logFeatureUsage } from '$lib/utils/featureUsage'
import { AssetService } from '$lib/gen'
import { PIPELINE_DRAFT_KIND, pipelineFolderFromBundlePath } from '$lib/pipelinePaths'
export type DeploymentStatus = { status: 'loading' | 'failed'; error?: string }
@@ -52,6 +54,9 @@ export interface SessionDeployModelArgs {
onItemDiscarded?: (item: DeployItem) => void
/** Chat-modified-items mask; undefined shows every draft. */
mask?: Set<string>
/** Deploys a pipeline folder's draft bundle, which no per-item deploy can: it
* expands into the folder's scripts and triggers. */
deployPipeline?: (folder: string) => Promise<DeployResult>
}
export interface DiffValues {
@@ -59,6 +64,15 @@ export interface DiffValues {
after: unknown
}
// A pipeline has no row of its own: it exists while its folder holds pipeline scripts.
async function itemExists(kind: Kind, path: string, workspace: string): Promise<boolean> {
if (kind !== PIPELINE_DRAFT_KIND) return checkItemExists(kind, path, workspace)
const folder = pipelineFolderFromBundlePath(path)
if (!folder) return false
const pipelines = await AssetService.listPipelineFolders({ workspace })
return pipelines.some((p) => p.folder === folder)
}
/** Resolves which mask-only candidates still exist in the session workspace
* (deployed) vs are gone (discarded drafts / deleted items) — the
* `existingKeys` input of `buildDeployItems`. Undefined while resolving, so
@@ -100,7 +114,7 @@ export function useExistingMaskKeys(
// workspace; a discarded or deleted one doesn't.
void Promise.all(
cands.map((c) =>
checkItemExists(c.deployKind, c.path, workspaceId)
itemExists(c.deployKind, c.path, workspaceId)
.then((exists) => (exists ? c.key : null))
.catch(() => null)
)
@@ -246,11 +260,18 @@ export function useSessionDeployModel(getArgs: () => SessionDeployModelArgs) {
async function runPlan(entry: DeployPlanEntry): Promise<DeployResult> {
const cur = getArgs().workspaceId
switch (entry.op) {
case 'deploy_draft':
case 'deploy_draft': {
const folder =
entry.draftKind === PIPELINE_DRAFT_KIND
? pipelineFolderFromBundlePath(entry.path)
: undefined
const deployPipeline = getArgs().deployPipeline
if (folder && deployPipeline) return deployPipeline(folder)
return deployDraft(entry.draftKind, entry.path, cur, {
draftOnly: entry.draftOnly,
rawApp: entry.rawApp
})
}
case 'discard':
return discardDraft(entry.draftKind, entry.path, cur, entry.draftOnly, entry.legacy)
}
@@ -314,6 +335,16 @@ export function useSessionDeployModel(getArgs: () => SessionDeployModelArgs) {
// ── Diff values (one resolver for tree + column) ─────────────────────────
async function loadDiffValues(item: DeployItem): Promise<DiffValues> {
const pipelineFolder =
item.draftKind === PIPELINE_DRAFT_KIND ? pipelineFolderFromBundlePath(item.path) : undefined
if (pipelineFolder) {
// Dynamic: the pipeline deploy module reaches every trigger kind's save utils.
const { loadPipelineDraftDiff } = await import(
'$lib/components/assets/AssetGraph/pipelineDeploy.svelte'
)
if (item.hasDraft) return loadPipelineDraftDiff(getArgs().workspaceId, pipelineFolder)
return { before: undefined, after: undefined }
}
const base = diffBaseFor(item, getArgs().workspaceId)
if (base.kind === 'draft') {
const { deployed, draft } = await getDraftDiffValues(
@@ -3,6 +3,8 @@ import { get } from 'svelte/store'
import { base } from '$lib/base'
import { AIChatManager, AIMode } from '$lib/components/copilot/chat/AIChatManager.svelte'
import { PipelineEditorState } from '$lib/components/assets/AssetGraph/pipelineEditorState.svelte'
import { pipelineLocalMirrorKey } from '$lib/pipelinePaths'
import type { DeployResult } from '$lib/utils_workspace_deploy'
import { initFlow } from '$lib/components/flows/flowStore.svelte'
import {
AppService,
@@ -129,6 +131,12 @@ export interface LoadSlot {
export type SessionTargetKind = 'flow' | 'script' | 'raw_app'
/** What a mounted pipeline editor view offers the rest of the session. */
export interface PipelineView {
/** Deploy all of the folder's drafts; failures stay drafts and are shown in the view. */
deployAll(): Promise<DeployResult>
}
// The live runtime value a raw-app editor cell binds. Legacy drag-and-drop apps
// are intentionally NOT hosted in the session preview (only code-based raw apps).
export interface RawAppRuntimeValue {
@@ -194,6 +202,14 @@ export interface SessionRuntime {
// Pipeline editor state per folder — persists across editor remounts and
// session switches, so it can't be component-local. Created on first use.
pipelineEditor(folder: string): PipelineEditorState
// A mounted pipeline editor view offers its deploy here; returns the unregister.
registerPipelineView(folder: string, view: PipelineView): () => void
// Deploy a folder's pipeline drafts through its editor tab, opening the tab if
// needed: the tab hydrates the drafts and shows what deployed and what failed.
deployPipeline(folder: string): Promise<DeployResult>
// Drop a folder's in-memory drafts once its draft bundle is discarded, so an
// open or later-mounted editor does not save them back.
forgetPipelineDrafts(folder: string): void
// Per-(kind, path) editor cells (content/baseline stores + load slot), created
// on demand. Each editable preview tab resolves its own cell, so several items
// stay live at once.
@@ -589,6 +605,61 @@ function createRuntime(session: Session): SessionRuntime {
return editor
}
const pipelineViews = new Map<string, PipelineView>()
const pipelineViewWaiters = new Map<string, Set<(view: PipelineView) => void>>()
function registerPipelineView(folder: string, view: PipelineView): () => void {
const key = normalizePipelineFolder(folder)
pipelineViews.set(key, view)
const waiters = pipelineViewWaiters.get(key)
pipelineViewWaiters.delete(key)
waiters?.forEach((notify) => notify(view))
return () => {
if (pipelineViews.get(key) === view) pipelineViews.delete(key)
}
}
function pipelineViewFor(folder: string, timeoutMs = 15000): Promise<PipelineView | undefined> {
const mounted = pipelineViews.get(folder)
if (mounted) return Promise.resolve(mounted)
return new Promise((resolve) => {
const waiters = pipelineViewWaiters.get(folder) ?? new Set()
pipelineViewWaiters.set(folder, waiters)
const notify = (view: PipelineView) => {
clearTimeout(timer)
resolve(view)
}
const timer = setTimeout(() => {
waiters.delete(notify)
resolve(undefined)
}, timeoutMs)
waiters.add(notify)
})
}
async function deployPipeline(folder: string): Promise<DeployResult> {
const key = normalizePipelineFolder(folder)
const target = previewTargetForSessionTarget('pipeline', key)
if (target) previewTabs.open(target)
const view = await pipelineViewFor(key)
if (!view) return { success: false, error: `The pipeline editor for f/${key} did not open.` }
return view.deployAll()
}
function forgetPipelineDrafts(folder: string): void {
const key = normalizePipelineFolder(folder)
const editor = pipelineEditors.get(key)
if (editor) {
editor.drafts = new Map()
editor.triggerDrafts = new Map()
editor.activeDraftPath = undefined
editor.clearLiveOverlays()
}
// The editor's crash mirror, which its load falls back to when the DB has
// no draft: left behind, it would restore what was just discarded.
try {
localStorage.removeItem(pipelineLocalMirrorKey(key))
} catch {
// Storage unavailable: nothing was mirrored either.
}
}
const runtimeLogRequesters = new Map<string, RawAppRuntimeLogRequester>()
// appPath → requester, one entry per mounted raw-app preview tab.
const domRequesters = new Map<string, RawAppDomRequester>()
@@ -602,6 +673,9 @@ function createRuntime(session: Session): SessionRuntime {
manager,
previewTabs,
pipelineEditor,
registerPipelineView,
deployPipeline,
forgetPipelineDrafts,
flowCell,
loadedEditorPath,
+5
View File
@@ -9,6 +9,11 @@ export function pipelineBundlePath(folder: string): string {
return `f/${folder}/${PIPELINE_DRAFT_KIND}`
}
/** localStorage key of the editor's crash mirror of a folder's bundle. */
export function pipelineLocalMirrorKey(folder: string): string {
return `pipeline-${folder}`
}
const BUNDLE_PATH_RE = new RegExp(`^f/([^/]+)/${PIPELINE_DRAFT_KIND}$`)
/** The folder a pipeline bundle belongs to, or `undefined` if the path isn't one. */
+2 -1
View File
@@ -779,7 +779,8 @@ const KIND_GATED_BY_DEPLOY_RULES: Record<Kind, boolean> = {
email_trigger: false,
datatable_migration: false,
trigger: false,
data_pipeline: false
// Deploying a pipeline deploys its scripts.
data_pipeline: true
}
// Every gated kind is spelled identically in `Kind` and `UserDraftItemKind`, so one lookup serves
@@ -94,10 +94,11 @@
isDraftableTriggerKind,
triggerDraftKey
} from '$lib/components/assets/AssetGraph/pipelineTriggerDrafts'
import { deleteTriggerRow } from '$lib/components/assets/AssetGraph/pipelineTriggerDraftDeploy'
import {
deleteTriggerRow,
deployTriggerDraft
} from '$lib/components/assets/AssetGraph/pipelineTriggerDraftDeploy'
deployPipelineDrafts,
deployPipelineScript
} from '$lib/components/assets/AssetGraph/pipelineDeploy.svelte'
import type { AssetUpstreamDelete } from '$lib/components/assets/AssetGraph/assetsOnlyView'
import type { NodeFixSpec } from '$lib/components/assets/AssetGraph/NodeFixButton.svelte'
import { TRIGGER_NODE_STYLE } from '$lib/components/assets/AssetGraph/TriggerNode.svelte'
@@ -117,6 +118,8 @@
} from '$lib/components/recording/pipelineRecording.svelte'
import type { PipelineRecording } from '$lib/components/recording/types'
import AutosaveIndicator from '$lib/components/AutosaveIndicator.svelte'
import PipelineDeployErrors from '$lib/components/assets/AssetGraph/PipelineDeployErrors.svelte'
import PipelineDeployTriggersModal from '$lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte'
import { onMount, tick, untrack } from 'svelte'
import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
import OpenInSessionButton, {
@@ -153,11 +156,9 @@
} from '$lib/gen'
import { resource } from 'runed'
import { emptySchema, sendUserToast, type Item } from '$lib/utils'
import type { Schema } from '$lib/common'
import { beforeNavigate, goto } from '$app/navigation'
import { fade } from 'svelte/transition'
import Popover from '$lib/components/meltComponents/Popover.svelte'
import { inferArgs, inferAssets } from '$lib/infer'
import { inferAssets } from '$lib/infer'
import PipelineTriggerEditors from '$lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte'
// Variables and resources are declarative config, not pipeline assets —
@@ -605,6 +606,9 @@
itemKind: PIPELINE_DRAFT_KIND,
path: pipelineDraftPath
})
// The drafts go with the pipeline into the session, so leaving for it is
// not leaving them behind.
bypassNavigationGuard = true
}
const sessionOpen: OpenInSessionSource | undefined = $derived(
@@ -745,54 +749,8 @@
if (msg) sendUserToast(msg, true)
}
async function saveDraft(path: string, draft: Draft, ws: string): Promise<void> {
const script = structuredClone($state.snapshot(draft.script) as Script)
script.schema = script.schema ?? emptySchema()
try {
const result = await inferArgs(script.language, script.content, script.schema as Schema)
;(script as any).auto_kind = result?.auto_kind || undefined
script.has_preprocessor = result?.has_preprocessor ?? false
} catch {
// Inference failures don't block deploys (the same fallback the
// per-pane save uses). The createScript call is the real
// validation gate — if the body is broken it'll reject there.
}
// Re-infer the asset lineage from the CURRENT body. The draft's
// `script.assets` is a snapshot that isn't refreshed on edit, so without
// this a renamed/removed output (e.g. an old `CREATE TABLE foo`) would
// be re-deployed as a phantom write edge and linger as an orphan asset.
// Mirrors the per-pane save, which sends live-inferred assets.
let assets: AssetWithAltAccessType[] = []
try {
const inferred = await inferAssets(script.language, script.content)
if (inferred?.status !== 'error')
assets = (inferred?.assets ?? []) as AssetWithAltAccessType[]
} catch {
// Same fallback as above — an unparsable body deploys with no
// lineage rather than the stale snapshot.
}
await ScriptService.createScript({
workspace: ws,
requestBody: {
...script,
language: script.language,
description: script.description ?? '',
// Brand-new drafts have no parent (buildDraft seeds hash as ''
// — treat any falsy hash as parentless, the backend rejects
// an empty hex string); a draft promoted from unsaved edits
// to a deployed script carries that script's hash and must
// chain off it (createScript rejects a parentless deploy to
// an occupied path).
parent_hash: script.hash ? String(script.hash) : undefined,
is_template: false,
tag: script.tag,
kind: script.kind as Script['kind'] | undefined,
lock: undefined,
// Freshly inferred above — overrides the stale snapshot carried
// by `...script`, so the deployed lineage matches the body.
assets: assets as any
}
})
function saveDraft(_path: string, draft: Draft, ws: string): Promise<void> {
return deployPipelineScript(draft, ws)
}
// Scripts plus trigger drafts: what "Save all" deploys.
@@ -804,110 +762,13 @@
else void saveAllDrafts()
}
// Runs after the script deploys: a trigger's `script_path` must already
// exist. Triggers whose script is still an undeployed draft stay drafts.
async function deployTriggerDrafts(
ws: string,
savedScriptPaths: string[],
errors: Map<string, string>
): Promise<string[]> {
const ready = [...pe.triggerDrafts].filter(
([, d]) =>
!pe.drafts.has(d.config.script_path) || savedScriptPaths.includes(d.config.script_path)
)
const results = await Promise.all(ready.map(([, d]) => deployTriggerDraft(d, ws)))
const saved: string[] = []
results.forEach((ok, i) => {
const [key, d] = ready[i]
if (ok) saved.push(key)
else errors.set(d.config.path, `Could not create the ${d.kind} trigger, see the notification`)
})
for (const key of saved) pe.discardTriggerDraft(key)
return saved
}
async function saveAllDrafts() {
if (!$workspaceStore || pendingCount === 0 || savingAll) return
savingAll = true
const ws = $workspaceStore
// 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))
const errors = new Map<string, string>()
const savedPaths: string[] = []
// Parallel — every createScript is independent. The backend handles
// its own ordering for any cross-script lock writes; we just want
// failures isolated per script so one bad body doesn't block the
// other deploys.
const results = await Promise.allSettled(
entries.map(async ([path, d]) => {
await saveDraft(path, d, ws)
return path
})
)
for (let i = 0; i < results.length; i++) {
const r = results[i]
const [path] = entries[i]
const msg =
r.status === 'rejected'
? String((r.reason as any)?.body ?? (r.reason as any)?.message ?? r.reason)
: ''
// A refused draft that equals the deployed script has nothing left to
// deploy. A duplicate refusal alone does not prove it: it is checked
// against every past version, not just the live one.
if (r.status === 'fulfilled') {
savedPaths.push(path)
} else if (await matchesDeployed(path, entries[i][1], ws)) {
savedPaths.push(path)
} else {
errors.set(
path,
/same hash/i.test(msg)
? 'This is the content of an earlier version, which cannot be deployed again as is. Change anything in it (a comment will do) to deploy it.'
: msg
)
}
}
const savedTriggers = await deployTriggerDrafts(ws, savedPaths, errors)
// Drop the saved drafts from the map; failed ones stay so the user
// can fix them and retry. Build the new map from the still-failing
// entries to keep insertion order stable.
if (savedPaths.length > 0) {
const next = new Map<string, Draft>()
for (const [k, v] of pe.drafts) {
if (!savedPaths.includes(k)) next.set(k, v)
}
pe.drafts = next
// If the open draft just got deployed, transfer the focus to
// its now-persisted runnable so the pane stays on the same
// script the user was editing — otherwise the pane closes,
// the canvas re-fits, and the user has to re-find their
// script after every save.
if (pe.activeDraftPath && savedPaths.includes(pe.activeDraftPath)) {
pe.selection = {
kind: 'runnable',
runnable_kind: 'script',
path: pe.activeDraftPath
}
pe.activeDraftPath = undefined
}
}
const predicted = predictCascadeFacts([...pe.drafts.keys()])
const { savedPaths, savedTriggers, errors } = await deployPipelineDrafts(pe, $workspaceStore)
if (savedPaths.length > 0 || savedTriggers.length > 0) {
await graphRes.refetch()
// Verify only what actually deployed — failed drafts would
@@ -1210,31 +1071,6 @@
// (same rationale as the live-callback handlers above) and so the
// template can gate them per-mode with simple ternaries — the canvas
// hides each affordance when its callback is undefined.
// Every field Save all deploys: a draft differing from the live script in any
// of them still has something to save.
const DEPLOYED_FIELDS = [
'content',
'language',
'summary',
'description',
'tag',
'kind',
'labels'
] as const
async function matchesDeployed(path: string, d: Draft, ws: string): Promise<boolean> {
try {
const live = await ScriptService.getScriptByPath({ workspace: ws, path })
const norm = (v: unknown) =>
JSON.stringify(Array.isArray(v) && v.length === 0 ? null : (v ?? null))
return DEPLOYED_FIELDS.every(
(f) =>
norm(f === 'description' ? (live[f] ?? '') : live[f]) ===
norm(f === 'description' ? (d.script[f] ?? '') : d.script[f])
)
} catch {
return false
}
}
function handleCanvasSelect(s: AssetGraphSelection | undefined) {
// Clicking a node while the pane is explicitly hidden is a request
// to see that node — unhide. Background clicks (s == undefined)
@@ -3231,44 +3067,7 @@
</Button>
{/if}
{#if mode === 'edit' && saveErrors.size > 0}
<!-- Compact errors popover anchored next to Save all so users
can see exactly which drafts failed and why without losing
the editor context. Drafts that succeed disappear from
the map; the ones still listed here are the unresolved
failures. -->
<Popover
placement="bottom-end"
contentClasses="p-3 max-w-[480px]"
usePointerDownOutside
enableFlyTransition
bind:isOpen={saveErrorsOpen}
>
{#snippet trigger()}
<button
type="button"
class="flex items-center gap-1.5 px-2 py-1 rounded-md text-red-600 dark:text-red-400 bg-red-50 dark:bg-red-900/30 hover:bg-red-100 dark:hover:bg-red-900/50 transition-colors text-xs font-medium"
title="View save errors"
>
<AlertTriangle size={14} />
<span>{saveErrors.size} failed</span>
</button>
{/snippet}
{#snippet content()}
<div class="flex flex-col gap-2">
<span class="text-xs font-semibold text-emphasis">Save errors</span>
<div class="flex flex-col gap-2 max-h-72 overflow-y-auto">
{#each [...saveErrors.entries()] as [path, message]}
<div class="flex flex-col gap-0.5 border-l-2 border-red-400 pl-2">
<span class="text-2xs font-mono text-emphasis">{path}</span>
<span class="text-2xs text-red-600 dark:text-red-400 break-words">
{message}
</span>
</div>
{/each}
</div>
</div>
{/snippet}
</Popover>
<PipelineDeployErrors errors={saveErrors} bind:open={saveErrorsOpen} />
{/if}
{#if mode === 'edit' && pendingCount > 0}
<Button
@@ -3605,11 +3404,9 @@
{/if}
</ConfirmationModal>
<ConfirmationModal
<PipelineDeployTriggersModal
open={confirmTriggerDeployOpen}
title="Deploy triggers?"
confirmationText="Deploy"
type="info"
triggerDrafts={pe.triggerDrafts}
onConfirmed={() => {
confirmTriggerDeployOpen = false
if (leaveAfterTriggerConfirm) {
@@ -3626,27 +3423,7 @@
pendingNavigationUrl = undefined
}
}}
>
<div class="flex flex-col gap-2 text-xs text-secondary">
<p>
{pe.triggerDrafts.size === 1
? 'Saving the pipeline also creates this trigger. It starts running its script once deployed.'
: `Saving the pipeline also creates these ${pe.triggerDrafts.size} triggers. They start running their scripts once deployed.`}
</p>
<ul class="flex flex-col gap-1">
{#each [...pe.triggerDrafts] as [key, d] (key)}
<li class="flex flex-col">
<span class="font-mono text-2xs text-emphasis truncate">{d.config.path}</span>
<span class="text-2xs text-hint truncate">
{d.kind}{#if d.kind === 'schedule'}
· <span class="font-mono">{d.config.schedule}</span> ({d.config.timezone}){/if}
→ {d.config.script_path}
</span>
</li>
{/each}
</ul>
</div>
</ConfirmationModal>
/>
<PipelineTriggerEditors
bind:this={triggerEditors}