diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
index a63bdca62b..53097f5379 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
@@ -323,8 +323,44 @@
 		// own adjacency (boundedCascade.buildLineageDownstreamMap).
 		const hasLineageDownstream = new Set<string>(buildLineageDownstreamMap(g).keys())
 
+		// Producers that declare `// data_test` checks, keyed by runnable id — the
+		// asset node uses this (plus the producer's run state) to render the
+		// data-test outcome badge. Tests only assert on ducklake `// materialize`
+		// targets (v1), so guard status is a ducklake-only concept below.
+		const producerHasTests = new Set<string>()
+		// Producer → its declared `// materialize` target, so a multi-output
+		// producer's guard badge lands only on the table its tests assert on —
+		// not on its other ducklake outputs. Mirrors the write-edge badge anchor.
+		const guardMaterializeTarget = new Map<
+			string,
+			NonNullable<AssetGraphResponse['runnables'][number]['materialize_target']>
+		>()
+		for (const r of g.runnables) {
+			if (r.data_tests && r.data_tests.length > 0) producerHasTests.add(`${r.usage_kind}:${r.path}`)
+			if (r.materialize_target)
+				guardMaterializeTarget.set(`${r.usage_kind}:${r.path}`, r.materialize_target)
+		}
+
 		for (const a of g.assets) {
 			const assetId = `asset:${a.kind}:${a.path}`
+			// Guard/outcome badge inputs: is a producer of this (ducklake) asset
+			// test-guarded, and did that guarded producer's latest run fail?
+			let dataTestGuarded = false
+			let producerFailed = false
+			if (a.kind === 'ducklake') {
+				for (const p of producersByAsset.get(`${a.kind}:${a.path}`) ?? []) {
+					const rid = `${p.kind}:${p.path}`
+					if (!producerHasTests.has(rid)) continue
+					// Tests assert on the producer's `// materialize` target; when the
+					// producer declares one, only that table is guarded (a multi-output
+					// producer must not badge its other ducklake writes). No declared
+					// target → single-output producer, so its lone ducklake write is it.
+					const mt = guardMaterializeTarget.get(rid)
+					if (mt && !(mt.kind === a.kind && mt.path === a.path)) continue
+					dataTestGuarded = true
+					if (runStates?.get(rid)?.status === 'failure') producerFailed = true
+				}
+			}
 			nodes.push({
 				id: assetId,
 				type: 'asset',
@@ -337,7 +373,9 @@
 					pathPrefix,
 					defaultPathSuffix,
 					producers: producersByAsset.get(`${a.kind}:${a.path}`) ?? [],
-					onRunProducer
+					onRunProducer,
+					dataTestGuarded,
+					producerFailed
 				}
 			})
 		}
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte
index 7d784b74b7..762bdbc0fc 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte
@@ -5,9 +5,19 @@
 	import { formatShortAssetPath, type AssetKind } from '$lib/components/assets/lib'
 	import { NODE } from '$lib/components/graph/util'
 	import PipelineInsertMenu, { type PipelineInsertPick } from './PipelineInsertMenu.svelte'
-	import { ArrowUpRight, Code2, GitFork, History, Play, Loader2, Plus } from 'lucide-svelte'
+	import {
+		ArrowUpRight,
+		Code2,
+		GitFork,
+		History,
+		Play,
+		Loader2,
+		Plus,
+		ShieldCheck,
+		ShieldAlert
+	} from 'lucide-svelte'
 	import type { ScriptLang } from '$lib/gen'
-	import { workspaceStore } from '$lib/stores'
+	import { enterpriseLicense, workspaceStore } from '$lib/stores'
 	import { sendUserToast } from '$lib/utils'
 	import { PIPELINE_LANGUAGES } from './pipelineLanguages'
 	import type { PipelineOutputKind } from './pipelineTemplates'
@@ -55,6 +65,15 @@
 			// cached draft content). Without this callback, the play button
 			// is hidden — runs only make sense in editor contexts.
 			onRunProducer?: (producer: AssetProducer) => Promise<string | undefined>
+			// True when a producer materializing this asset declares `// data_test`
+			// checks — the asset's write is guarded. Drives the data-test outcome
+			// badge, whose meaning differs by edition (EE rolls a failing write
+			// back; CE publishes it anyway).
+			dataTestGuarded?: boolean
+			// True when the latest observed run of a producer materializing this
+			// asset failed. Escalates the guard badge from "protected" to a
+			// failed-run outcome (rolled-back on EE, published-anyway on CE).
+			producerFailed?: boolean
 		}
 		// SvelteFlow injects this on the node component when the user clicks
 		// the node. Combined with our own `hovered` state to drive the
@@ -109,6 +128,37 @@
 	}
 
 	let showAdd = $derived(data.onAddScript != undefined)
+
+	// Data-test outcome badge. Only guarded assets show it. The write's fate on a
+	// failing test differs by edition — surface which one applies so a shared
+	// parent/fork table name can't hide a silently-published bad version.
+	let isEE = $derived(!!$enterpriseLicense)
+	let showGuardBadge = $derived(data.dataTestGuarded === true)
+	let guardFailed = $derived(data.producerFailed === true)
+	let GuardIcon = $derived(isEE ? ShieldCheck : ShieldAlert)
+	// Filled + colored when the last run failed (the actionable state); a quiet
+	// ring at rest so the badge doesn't shout on every healthy guarded asset.
+	let guardClass = $derived(
+		guardFailed
+			? isEE
+				? 'bg-amber-500 text-white border-amber-600'
+				: 'bg-red-500 text-white border-red-600'
+			: isEE
+				? 'bg-surface-secondary text-emerald-600 dark:text-emerald-400 border-emerald-500/60'
+				: 'bg-surface-secondary text-amber-600 dark:text-amber-400 border-amber-500/60'
+	)
+	// Failed copy speaks to the edition's write policy, not the failure cause:
+	// `producerFailed` is a generic job failure (could be a runtime/worker error,
+	// not a data-test violation), so we don't assert "failed its data tests".
+	let guardTitle = $derived(
+		guardFailed
+			? isEE
+				? 'Last run failed — Enterprise rolls a failed materialize back, so the previous version is left live.'
+				: 'Last run failed — Community Edition does not roll a failed materialize back (Enterprise does), so a failing write may be left live. Verify the table.'
+			: isEE
+				? 'Guarded by data tests: a failing write is rolled back, keeping the previous version live.'
+				: 'Guarded by data tests, but Community Edition does not block on failure — a failing write is still published. Rollback is Enterprise-only.'
+	)
 </script>
 
 <!-- onmouseenter/leave on the wrapper (not the inner card) so the run
@@ -142,23 +192,26 @@
 		<span class="flex-1 min-w-0 pr-1 py-0.5 text-2xs font-mono text-emphasis truncate">
 			{formatShortAssetPath(asset)}
 		</span>
-		<!-- Fork data-environment marker: amber ↗ = deferred (reads the parent
-		     workspace's current table via a view), emerald fork glyph = the fork
-		     materialized its own copy. Icon-only — the pill truncates its path
-		     already, a labeled chip wouldn't fit; the title carries the meaning. -->
+		<!-- Fork data-environment chip: in a fork every asset shares its parent's
+		     name, so the env it resolves to must read at a glance. Labeled + tinted
+		     (amber "parent" = deferred read of the parent's current table via a
+		     view; emerald "fork" = the fork's own materialized copy) rather than a
+		     bare icon, which was too easy to miss. The title carries the detail. -->
 		{#if data.fork_materialization === 'deferred'}
 			<span
-				class="shrink-0 mr-1.5 text-amber-600 dark:text-amber-400"
+				class="shrink-0 mr-1.5 flex items-center gap-0.5 rounded px-1 py-px text-3xs font-semibold uppercase tracking-wide bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300 border border-amber-300 dark:border-amber-700"
 				title="Deferred to parent workspace: reads the parent's current data. Materialize it in this fork to iterate on it."
 			>
-				<ArrowUpRight size={12} />
+				<ArrowUpRight size={10} />
+				parent
 			</span>
 		{:else if data.fork_materialization === 'fork'}
 			<span
-				class="shrink-0 mr-1.5 text-emerald-600 dark:text-emerald-500"
+				class="shrink-0 mr-1.5 flex items-center gap-0.5 rounded px-1 py-px text-3xs font-semibold uppercase tracking-wide bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300 border border-emerald-300 dark:border-emerald-700"
 				title="Materialized in this fork: reads and writes use the fork's isolated copy."
 			>
-				<GitFork size={12} />
+				<GitFork size={10} />
+				fork
 			</span>
 		{/if}
 		<!-- SCD2 companion marker: this node is the `<dim>_current` "latest row
@@ -173,6 +226,21 @@
 			</span>
 		{/if}
 	</div>
+	{#if showGuardBadge}
+		<!-- Data-test outcome badge. Floats off the TOP-RIGHT corner (opposite the
+		     left-edge run button and the bottom + inserter) so it never collides
+		     with the other node affordances. Shield = guarded; its fill escalates to
+		     amber/red when the last run failed, encoding the edition's write policy. -->
+		<div
+			class={twMerge(
+				'absolute -top-2 -right-2 z-10 rounded-full w-5 h-5 grid place-items-center border shadow-sm',
+				guardClass
+			)}
+			title={guardTitle}
+		>
+			<GuardIcon size={12} strokeWidth={2.25} />
+		</div>
+	{/if}
 	{#if showActions}
 		<!-- Run button revealed on hover/select. Floats off the LEFT edge —
 		     visually closer to the upstream producer it triggers (which lays
diff --git a/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.test.ts b/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.test.ts
index 92d7a977fe..3a0fd4fbd3 100644
--- a/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.test.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.test.ts
@@ -201,7 +201,48 @@ describe('runSelection', () => {
 		expect(r.launched.indexOf('c')).toBeGreaterThan(r.launched.indexOf('b'))
 	})
 
-	it('stops scheduling after a failure', async () => {
+	it('skips a failed node’s descendants but keeps independent branches running', async () => {
+		// Two independent chains: a → b and c → d. `a` fails; `b` (its descendant)
+		// must be skipped, but `d` depends only on the successful `c`, so it must
+		// still run — a failure must not stall unrelated branches.
+		const sched = schedule(
+			[
+				['a', 'b'],
+				['c', 'd']
+			],
+			['a', 'b', 'c', 'd'],
+			['a', 'c']
+		)
+		const r = fakeRunner({ a: 'failure' })
+		const res = await runSelection({ schedule: sched, ...r })
+		expect(res.ok).toBe(false)
+		expect(res.statuses.get('a')?.status).toBe('failure')
+		expect(res.statuses.get('b')?.status).toBe('skipped')
+		expect(res.statuses.get('c')?.status).toBe('success')
+		expect(res.statuses.get('d')?.status).toBe('success')
+		expect(r.launched).toContain('d')
+		expect(r.launched).not.toContain('b')
+	})
+
+	it('skips a join node when any one of its upstreams fails', async () => {
+		// {a, b} → c. `a` fails; `c` needs both, so it must be skipped even though
+		// `b` succeeds — a poisoned lineage isn’t rescued by a sibling success.
+		const sched = schedule(
+			[
+				['a', 'c'],
+				['b', 'c']
+			],
+			['a', 'b', 'c'],
+			['a', 'b']
+		)
+		const r = fakeRunner({ a: 'failure' })
+		const res = await runSelection({ schedule: sched, ...r })
+		expect(res.ok).toBe(false)
+		expect(res.statuses.get('c')?.status).toBe('skipped')
+		expect(r.launched).not.toContain('c')
+	})
+
+	it('stops scheduling a failed node’s chain', async () => {
 		const sched = schedule([['a', 'b']], ['a', 'b'], ['a'])
 		const r = fakeRunner({ a: 'failure' })
 		const res = await runSelection({ schedule: sched, ...r })
diff --git a/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.ts b/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.ts
index 16d357af00..424f1581ec 100644
--- a/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/cascadeOrchestrator.ts
@@ -128,8 +128,9 @@ export type SelectionRunOptions = {
  * Execute an arbitrary selected set of scripts (e.g. a bounded-cascade
  * selection) in topological order. Unlike `runCascade` there is no single
  * privileged root — every `schedule.roots` entry is seeded at once and a node
- * runs as soon as its in-set upstreams all succeed. Failure stops *scheduling*
- * (in-flight jobs finish); everything not yet started ends 'skipped'.
+ * runs as soon as its in-set upstreams all succeed. A failure abandons only
+ * that node's lineage (its transitive descendants end 'skipped'); INDEPENDENT
+ * branches keep running. In-flight jobs always finish.
  */
 export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRunResult> {
 	const { schedule, launch, waitTerminal, onUpdate } = opts
@@ -137,10 +138,29 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
 	for (const n of schedule.nodes) statuses.set(n, { status: 'pending' })
 	const remaining = new Map(schedule.indegree)
 	let failed = false
+	// Nodes a failed prerequisite has made unrunnable — the transitive descendants
+	// of every failure. Gating scheduling on this (rather than a single global
+	// fail-fast flag) is what lets independent branches finish: only the lineage
+	// below a failure is skipped, not every node that happens to become ready
+	// afterwards. Poisoned nodes are never scheduled, so they end 'skipped' below.
+	const poisoned = new Set<string>()
 	const inFlight = new Set<Promise<void>>()
 
 	const emit = () => onUpdate?.(new Map(statuses))
 
+	function poison(path: string) {
+		const stack = [path]
+		while (stack.length > 0) {
+			const n = stack.pop()!
+			for (const s of schedule.edges.get(n) ?? []) {
+				if (!poisoned.has(s)) {
+					poisoned.add(s)
+					stack.push(s)
+				}
+			}
+		}
+	}
+
 	function schedule_(path: string) {
 		const p = runNode(path).finally(() => inFlight.delete(p))
 		inFlight.add(p)
@@ -159,6 +179,7 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
 			emit()
 			if (term === 'failure') {
 				failed = true
+				poison(path)
 				return
 			}
 		} catch (e) {
@@ -169,12 +190,13 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
 			})
 			emit()
 			failed = true
+			poison(path)
 			return
 		}
 		for (const s of schedule.edges.get(path) ?? []) {
 			const d = (remaining.get(s) ?? 0) - 1
 			remaining.set(s, d)
-			if (d === 0 && !failed) schedule_(s)
+			if (d === 0 && !poisoned.has(s)) schedule_(s)
 		}
 	}
 
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
index 167aaa8f25..d09c1be15f 100644
--- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
@@ -1643,6 +1643,25 @@
 		}
 	}
 
+	// Every pipeline-member script, for the always-visible header "Run pipeline"
+	// control. Per-node runs are hover/select-gated on the canvas; this
+	// pipeline-level affordance runs the whole graph without hunting for a root
+	// node to hover. Gated on `in_pipeline` so it never launches dependency-only
+	// endpoints the graph surfaces for context (macro libraries, custom
+	// data-test scripts, out-of-folder producers) — only actual pipeline steps.
+	let allPipelineScripts = $derived(
+		displayGraph.runnables
+			.filter((r) => r.usage_kind === 'script' && r.in_pipeline)
+			.map((r) => r.path)
+	)
+	// Run the whole pipeline: hand every script to the bounded-cascade engine,
+	// which topo-orders them (roots first) and fans downstream — i.e. run + all
+	// downstream from every source at once. Reuses the same per-hop launch/poll
+	// and one-cascade-at-a-time guard as the node-level chain runs.
+	async function runWholePipeline() {
+		await runBoundedCascade(allPipelineScripts)
+	}
+
 	// Counter bumped when the canvas Run button targets the currently-open
 	// script — the pane intercepts and routes through ScriptEditor.runTest
 	// so logs/result/cancel land in the test panel instead of going off
@@ -2004,6 +2023,18 @@
 		return `${n} ${singular}${n === 1 ? '' : 's'}`
 	}
 
+	// Human noun per asset kind for the header summary. The raw `ducklake` kind
+	// counts materialized tables/views (including `_current` history views), so
+	// "N ducklakes" mis-reads as a count of lakes — surface "table" instead. Other
+	// kinds keep their own noun; unmapped kinds fall back to the raw kind.
+	const ASSET_KIND_NOUN: Record<string, string> = {
+		ducklake: 'table',
+		datatable: 'table',
+		s3object: 'file',
+		volume: 'volume',
+		resource: 'resource'
+	}
+
 	let summary = $derived.by<string[]>(() => {
 		const g = graphRes.current
 		if (!g) return []
@@ -2012,9 +2043,14 @@
 		const flows = g.runnables.filter((r) => r.usage_kind === 'flow').length
 		if (scripts) parts.push(pluralize(scripts, 'script'))
 		if (flows) parts.push(pluralize(flows, 'flow'))
-		const byKind = new Map<string, number>()
-		for (const a of g.assets) byKind.set(a.kind, (byKind.get(a.kind) ?? 0) + 1)
-		for (const [kind, n] of byKind) parts.push(pluralize(n, kind))
+		// Collapse kinds that share a noun (ducklake + datatable → "table") into a
+		// single tally so the summary reads "5 tables", not "3 tables · 2 tables".
+		const byNoun = new Map<string, number>()
+		for (const a of g.assets) {
+			const noun = ASSET_KIND_NOUN[a.kind] ?? a.kind
+			byNoun.set(noun, (byNoun.get(noun) ?? 0) + 1)
+		}
+		for (const [noun, n] of byNoun) parts.push(pluralize(n, noun))
 		return parts
 	})
 </script>
@@ -2098,6 +2134,28 @@
 			</div>
 		{/if}
 		<div class="flex flex-row items-center gap-2 flex-1 justify-end">
+			{#if mode === 'edit' && !isOperator && allPipelineScripts.length > 0}
+				<!-- Pipeline-level run: always visible so a run doesn't require
+				     hovering a specific node's play button. Runs every script in
+				     dependency order (roots first, cascading downstream). -->
+				<Button
+					variant="accent-secondary"
+					unifiedSize="sm"
+					startIcon={{
+						icon: cascadeRunningRoot ? Loader2 : Play,
+						classes: cascadeRunningRoot ? 'animate-spin' : undefined
+					}}
+					onclick={runWholePipeline}
+					disabled={!!cascadeRunningRoot}
+					title={cascadeRunningRoot
+						? 'A pipeline run is already in progress'
+						: `Run all ${allPipelineScripts.length} script${
+								allPipelineScripts.length === 1 ? '' : 's'
+							} in dependency order (roots first, cascading downstream)`}
+				>
+					{cascadeRunningRoot ? 'Running…' : 'Run pipeline'}
+				</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