From dedffeb04647ce4eb4ad11e71bcf7d04d0bda7cf Mon Sep 17 00:00:00 2001
From: Diego Imbert <diego@windmill.dev>
Date: Wed, 30 Sep 2026 20:39:57 +0200
Subject: [PATCH] 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>
---
 .../AssetGraph/PipelineDeployErrors.svelte    |  49 ++++
 .../PipelineDeployTriggersModal.svelte        |  47 ++++
 .../AssetGraph/PipelineGraphEditor.svelte     |   8 +-
 .../AssetGraph/pipelineDeploy.svelte.ts       | 205 ++++++++++++++
 .../copilot/chat/pipeline/core.test.ts        |   8 +-
 .../components/copilot/chat/pipeline/core.ts  |  12 +-
 .../sessions/PipelineEditorView.svelte        | 129 ++++++++-
 .../sessions/SessionChangesBar.svelte         |   9 +-
 .../sessions/SessionDiffDrawer.svelte         |   9 +-
 .../sessions/modifiedItemsMask.test.ts        |   7 +
 .../components/sessions/modifiedItemsMask.ts  |  10 +
 .../sessions/sessionDeployModel.svelte.ts     |  35 ++-
 .../sessions/sessionRuntime.svelte.ts         |  74 +++++
 frontend/src/lib/pipelinePaths.ts             |   5 +
 frontend/src/lib/utils_workspace_deploy.ts    |   3 +-
 .../(logged)/pipeline/[folder]/+page.svelte   | 259 ++----------------
 16 files changed, 616 insertions(+), 253 deletions(-)
 create mode 100644 frontend/src/lib/components/assets/AssetGraph/PipelineDeployErrors.svelte
 create mode 100644 frontend/src/lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte
 create mode 100644 frontend/src/lib/components/assets/AssetGraph/pipelineDeploy.svelte.ts

diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineDeployErrors.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineDeployErrors.svelte
new file mode 100644
index 0000000000..6206ebdf33
--- /dev/null
+++ b/frontend/src/lib/components/assets/AssetGraph/PipelineDeployErrors.svelte
@@ -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>
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte
new file mode 100644
index 0000000000..d96b758625
--- /dev/null
+++ b/frontend/src/lib/components/assets/AssetGraph/PipelineDeployTriggersModal.svelte
@@ -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>
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
index 83b098a6f7..8072e25ec1 100644
--- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
@@ -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
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineDeploy.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineDeploy.svelte.ts
new file mode 100644
index 0000000000..26ddff6b31
--- /dev/null
+++ b/frontend/src/lib/components/assets/AssetGraph/pipelineDeploy.svelte.ts
@@ -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 }
+}
diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts
index c40bdeb015..dd5579eb45 100644
--- a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts
+++ b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts
@@ -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',
diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.ts
index b1bad57a7e..efc734373f 100644
--- a/frontend/src/lib/components/copilot/chat/pipeline/core.ts
+++ b/frontend/src/lib/components/copilot/chat/pipeline/core.ts
@@ -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'
diff --git a/frontend/src/lib/components/sessions/PipelineEditorView.svelte b/frontend/src/lib/components/sessions/PipelineEditorView.svelte
index 930154505f..7b3e646b42 100644
--- a/frontend/src/lib/components/sessions/PipelineEditorView.svelte
+++ b/frontend/src/lib/components/sessions/PipelineEditorView.svelte
@@ -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)}
+/>
diff --git a/frontend/src/lib/components/sessions/SessionChangesBar.svelte b/frontend/src/lib/components/sessions/SessionChangesBar.svelte
index 777ca8c935..fdec57df99 100644
--- a/frontend/src/lib/components/sessions/SessionChangesBar.svelte
+++ b/frontend/src/lib/components/sessions/SessionChangesBar.svelte
@@ -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}
diff --git a/frontend/src/lib/components/sessions/SessionDiffDrawer.svelte b/frontend/src/lib/components/sessions/SessionDiffDrawer.svelte
index fe5ab2d199..3bfc392999 100644
--- a/frontend/src/lib/components/sessions/SessionDiffDrawer.svelte
+++ b/frontend/src/lib/components/sessions/SessionDiffDrawer.svelte
@@ -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).
diff --git a/frontend/src/lib/components/sessions/modifiedItemsMask.test.ts b/frontend/src/lib/components/sessions/modifiedItemsMask.test.ts
index 6cad6121cd..a75d8a1803 100644
--- a/frontend/src/lib/components/sessions/modifiedItemsMask.test.ts
+++ b/frontend/src/lib/components/sessions/modifiedItemsMask.test.ts
@@ -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)
diff --git a/frontend/src/lib/components/sessions/modifiedItemsMask.ts b/frontend/src/lib/components/sessions/modifiedItemsMask.ts
index 7d0d3abbae..9885eceb1c 100644
--- a/frontend/src/lib/components/sessions/modifiedItemsMask.ts
+++ b/frontend/src/lib/components/sessions/modifiedItemsMask.ts
@@ -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
diff --git a/frontend/src/lib/components/sessions/sessionDeployModel.svelte.ts b/frontend/src/lib/components/sessions/sessionDeployModel.svelte.ts
index 981b4cc752..a99d38ee96 100644
--- a/frontend/src/lib/components/sessions/sessionDeployModel.svelte.ts
+++ b/frontend/src/lib/components/sessions/sessionDeployModel.svelte.ts
@@ -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(
diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts
index 9c1296c0ad..766690ca7d 100644
--- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts
+++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts
@@ -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,
 
diff --git a/frontend/src/lib/pipelinePaths.ts b/frontend/src/lib/pipelinePaths.ts
index 534e691479..d12b162e06 100644
--- a/frontend/src/lib/pipelinePaths.ts
+++ b/frontend/src/lib/pipelinePaths.ts
@@ -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. */
diff --git a/frontend/src/lib/utils_workspace_deploy.ts b/frontend/src/lib/utils_workspace_deploy.ts
index ba1f293678..280d0daa00 100644
--- a/frontend/src/lib/utils_workspace_deploy.ts
+++ b/frontend/src/lib/utils_workspace_deploy.ts
@@ -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
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
index bd18ef73f1..3cffc37443 100644
--- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
@@ -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}