From f84c4cf9630ec49d3aac6f1195318aaaa002ea81 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Tue, 16 Jun 2026 17:21:23 +0200
Subject: [PATCH] =?UTF-8?q?feat:=20pipeline=20Activity=20panel=20grouping,?=
 =?UTF-8?q?=20run=E2=86=94graph=20highlight,=20deploy-conflict=20handling?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Activity panel (view mode):
- Group cascade runs by the connected component of the asset-dispatch graph
  (new GET /jobs/asset_dispatch_edges over the dispatch_event table, incl.
  join_pending inputs), headed by the earliest originating run + its trigger,
  with a "+N" chip for joins fed by multiple triggers.
- Success/failure count histogram with drag-to-filter brushing, an always-on
  time axis + per-bar tooltips, a Reset, and Last hour/24h/48h/7/30/90d ranges.
- Node run-count/status badges now derive from the same merged historic+live
  events the panel shows (previously session-only).

Run ↔ graph highlight:
- Hovering a run row (or a group header → the whole cascade) rings the
  node(s), animates their incident edges, and borders the adjacent assets in
  the edge hue (blue write / gray read); expanding a run pins a soft-blue ring.
- Switching edit→view re-surfaces the Activity feed.

Deploy:
- Live-content autosave for the open pipeline draft + an autosave indicator.
- Re-saving a script now chains off the hash just created instead of a stale
  parent_hash (fixes the "lineage must be linear" error on a second save), and
  a genuine concurrent deploy opens a keep-mine / view-latest conflict modal.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 ...3460b6438d0a320b5554acec0beddbdeb008b.json |  74 ++++
 backend/windmill-api/openapi.yaml             |  72 ++++
 backend/windmill-api/src/jobs.rs              |  81 +++++
 .../AssetGraph/ActivityHistogram.svelte       | 155 ++++++++
 .../assets/AssetGraph/AssetGraphCanvas.svelte |  84 ++++-
 .../AssetGraph/AssetGraphDetailsPane.svelte   | 130 ++++++-
 .../AssetGraph/PipelineActivityPanel.svelte   | 331 +++++++++++++++++-
 .../AssetGraph/pipelineHistory.svelte.ts      |  34 ++
 .../(logged)/pipeline/[folder]/+page.svelte   | 128 ++++++-
 9 files changed, 1056 insertions(+), 33 deletions(-)
 create mode 100644 backend/.sqlx/query-8cd02a5378bea03012e6cb937f23460b6438d0a320b5554acec0beddbdeb008b.json
 create mode 100644 frontend/src/lib/components/assets/AssetGraph/ActivityHistogram.svelte

diff --git a/backend/.sqlx/query-8cd02a5378bea03012e6cb937f23460b6438d0a320b5554acec0beddbdeb008b.json b/backend/.sqlx/query-8cd02a5378bea03012e6cb937f23460b6438d0a320b5554acec0beddbdeb008b.json
new file mode 100644
index 0000000000..24d70efae1
--- /dev/null
+++ b/backend/.sqlx/query-8cd02a5378bea03012e6cb937f23460b6438d0a320b5554acec0beddbdeb008b.json
@@ -0,0 +1,74 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n              de.producer_job_id AS \"producer_job_id!\",\n              de.child_job_id,\n              de.subscriber_path AS \"subscriber_path!\",\n              de.outcome::text AS \"outcome!\",\n              de.asset_kind AS \"asset_kind!: windmill_common::assets::AssetKind\",\n              de.asset_path AS \"asset_path!\",\n              de.created_at AS \"created_at!\"\n           FROM dispatch_event de\n           JOIN v2_job pj ON pj.id = de.producer_job_id\n           WHERE de.workspace_id = $1\n             AND de.outcome IN ('dispatched', 'join_pending')\n             AND de.subscriber_path LIKE $2\n             AND ($3::timestamptz IS NULL OR de.created_at >= $3)\n           ORDER BY de.created_at DESC, de.id DESC\n           LIMIT 4000",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "producer_job_id!",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "child_job_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "subscriber_path!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "outcome!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "asset_kind!: windmill_common::assets::AssetKind",
+        "type_info": {
+          "Custom": {
+            "name": "asset_kind",
+            "kind": {
+              "Enum": [
+                "s3object",
+                "resource",
+                "variable",
+                "ducklake",
+                "datatable",
+                "volume"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 5,
+        "name": "asset_path!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 6,
+        "name": "created_at!",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Timestamptz"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      false,
+      null,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "8cd02a5378bea03012e6cb937f23460b6438d0a320b5554acec0beddbdeb008b"
+}
diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml
index 8cc9efd2ef..e99a1917dc 100644
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -13006,6 +13006,78 @@ paths:
                     - outcome
                     - created_at
 
+  /w/{workspace}/jobs/asset_dispatch_edges:
+    get:
+      summary: list asset-cascade producer→child job edges for a folder
+      description: >
+        Returns the `dispatched` asset-trigger edges (producer job → child job)
+        whose subscriber lives under `path_start`. Lets a pipeline view
+        reconstruct the cascade tree of a folder by job id and group connected
+        runs. Visibility follows the producer job's RLS.
+      operationId: listAssetDispatchEdges
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: path_start
+          in: query
+          required: true
+          description: Folder path prefix the children live under, e.g. `f/orders/`.
+          schema:
+            type: string
+        - name: created_after
+          in: query
+          required: false
+          description: Only edges dispatched at/after this instant.
+          schema:
+            type: string
+            format: date-time
+      responses:
+        "200":
+          description: asset-cascade edges for the folder
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    producer_job_id:
+                      type: string
+                      format: uuid
+                    child_job_id:
+                      type: string
+                      format: uuid
+                      description: Set for `dispatched`; absent for `join_pending` inputs.
+                    subscriber_path:
+                      type: string
+                    outcome:
+                      type: string
+                      enum:
+                        - dispatched
+                        - join_pending
+                    asset_kind:
+                      type: string
+                      enum:
+                        - s3object
+                        - resource
+                        - variable
+                        - ducklake
+                        - datatable
+                        - volume
+                    asset_path:
+                      type: string
+                    created_at:
+                      type: string
+                      format: date-time
+                  required:
+                    - producer_job_id
+                    - subscriber_path
+                    - outcome
+                    - asset_kind
+                    - asset_path
+                    - created_at
+
   /w/{workspace}/jobs/completed/delete/{id}:
     post:
       summary: delete completed job (erase content but keep run id)
diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs
index a9e307fc6e..d10abdf43e 100644
--- a/backend/windmill-api/src/jobs.rs
+++ b/backend/windmill-api/src/jobs.rs
@@ -270,6 +270,7 @@ pub fn workspaced_service() -> Router {
         )
         .route("/run/dynamic_select", post(run_dynamic_select))
         .route("/list", get(list_jobs))
+        .route("/asset_dispatch_edges", get(list_asset_dispatch_edges))
         .route(
             "/list_selected_job_groups",
             // We use post because sending a huge array as a query param can produce
@@ -9145,6 +9146,86 @@ async fn get_dispatch_events(
     ))
 }
 
+/// One asset-cascade dispatch record, for reconstructing the cascade graph of a
+/// pipeline folder in the Activity panel. `dispatched` rows carry the resolved
+/// `child_job_id` (a real producer→child job edge); `join_pending` rows are the
+/// pre-completion inputs of an AND-join (no child yet) — the client links them
+/// to the eventual child of the same `subscriber_path` so a join's separate
+/// trigger chains merge into one group. `skipped` is omitted.
+#[derive(Serialize)]
+struct AssetDispatchEdge {
+    producer_job_id: Uuid,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    child_job_id: Option<Uuid>,
+    subscriber_path: String,
+    outcome: String,
+    asset_kind: windmill_common::assets::AssetKind,
+    asset_path: String,
+    created_at: chrono::DateTime<Utc>,
+}
+
+#[derive(Deserialize)]
+struct AssetDispatchEdgesQuery {
+    /// Folder path prefix the children live under, e.g. `f/orders/`. Matched
+    /// against `subscriber_path` — every intra-pipeline cascade edge has its
+    /// child in the folder, so this captures the whole folder's cascades.
+    path_start: String,
+    /// Only edges dispatched at/after this instant (align with the activity
+    /// window the client already loaded). Omit for the default cap.
+    created_after: Option<chrono::DateTime<Utc>>,
+}
+
+/// Asset-cascade edges for a pipeline folder. RLS on the joined `v2_job`
+/// producer row limits this to cascades whose producer the caller can already
+/// see (same visibility as the folder's job list).
+async fn list_asset_dispatch_edges(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+    Query(query): Query<AssetDispatchEdgesQuery>,
+) -> error::JsonResult<Vec<AssetDispatchEdge>> {
+    let like = format!("{}%", query.path_start);
+    let mut tx = user_db.begin(&authed).await?;
+    let rows = sqlx::query!(
+        r#"SELECT
+              de.producer_job_id AS "producer_job_id!",
+              de.child_job_id,
+              de.subscriber_path AS "subscriber_path!",
+              de.outcome::text AS "outcome!",
+              de.asset_kind AS "asset_kind!: windmill_common::assets::AssetKind",
+              de.asset_path AS "asset_path!",
+              de.created_at AS "created_at!"
+           FROM dispatch_event de
+           JOIN v2_job pj ON pj.id = de.producer_job_id
+           WHERE de.workspace_id = $1
+             AND de.outcome IN ('dispatched', 'join_pending')
+             AND de.subscriber_path LIKE $2
+             AND ($3::timestamptz IS NULL OR de.created_at >= $3)
+           ORDER BY de.created_at DESC, de.id DESC
+           LIMIT 4000"#,
+        &w_id,
+        like,
+        query.created_after,
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    tx.commit().await?;
+
+    Ok(Json(
+        rows.into_iter()
+            .map(|r| AssetDispatchEdge {
+                producer_job_id: r.producer_job_id,
+                child_job_id: r.child_job_id,
+                subscriber_path: r.subscriber_path,
+                outcome: r.outcome,
+                asset_kind: r.asset_kind,
+                asset_path: r.asset_path,
+                created_at: r.created_at,
+            })
+            .collect(),
+    ))
+}
+
 async fn get_completed_job_timing(
     OptViewToken(view_token): OptViewToken,
     OptAuthed(opt_authed): OptAuthed,
diff --git a/frontend/src/lib/components/assets/AssetGraph/ActivityHistogram.svelte b/frontend/src/lib/components/assets/AssetGraph/ActivityHistogram.svelte
new file mode 100644
index 0000000000..fcfd95a5d3
--- /dev/null
+++ b/frontend/src/lib/components/assets/AssetGraph/ActivityHistogram.svelte
@@ -0,0 +1,155 @@
+<script lang="ts">
+	import { isActiveEvent, type PipelineEvent } from './activeRunnables.svelte'
+
+	interface Props {
+		// All runs in the loaded window — bucketed into the bars. Filtering is
+		// the caller's job; this always shows the whole window so you can brush
+		// within it.
+		events: PipelineEvent[]
+		// Window bounds in ms (start … end). Bars span this range.
+		from: number
+		to: number
+		// Current brushed selection in ms, or undefined for the whole window.
+		selected?: { from: number; to: number } | undefined
+		// Emitted on brush end; `undefined` for a click (clears the selection).
+		onSelect: (range: { from: number; to: number } | undefined) => void
+	}
+	let { events, from, to, selected, onSelect }: Props = $props()
+
+	const BUCKETS = 48
+
+	let bins = $derived.by(() => {
+		const span = Math.max(1, to - from)
+		const arr = Array.from({ length: BUCKETS }, () => ({ success: 0, failure: 0 }))
+		for (const e of events) {
+			// Future-scheduled queued runs aren't activity.
+			if (e.status === 'queued' && !isActiveEvent(e)) continue
+			const t = Date.parse(e.at)
+			if (isNaN(t) || t < from || t > to) continue
+			const i = Math.min(BUCKETS - 1, Math.max(0, Math.floor(((t - from) / span) * BUCKETS)))
+			if (e.status === 'failure') arr[i].failure += 1
+			else arr[i].success += 1 // success + running/active
+		}
+		return arr
+	})
+	let maxCount = $derived(Math.max(1, ...bins.map((b) => b.success + b.failure)))
+
+	// Selection rendered as a fractional band [0,1] over the bar area.
+	let selFrac = $derived.by(() => {
+		if (!selected) return undefined
+		const span = Math.max(1, to - from)
+		const a = Math.max(0, Math.min(1, (selected.from - from) / span))
+		const b = Math.max(0, Math.min(1, (selected.to - from) / span))
+		return a < b ? { left: a, width: b - a } : undefined
+	})
+
+	let barsEl: HTMLDivElement | undefined = $state()
+	let dragFrom = $state<number | undefined>(undefined) // fractional x at pointerdown
+	let dragTo = $state<number | undefined>(undefined)
+
+	function fracAt(clientX: number): number {
+		if (!barsEl) return 0
+		const r = barsEl.getBoundingClientRect()
+		return Math.max(0, Math.min(1, (clientX - r.left) / Math.max(1, r.width)))
+	}
+	function fracToMs(f: number): number {
+		return from + f * Math.max(1, to - from)
+	}
+	function onPointerDown(e: PointerEvent) {
+		barsEl?.setPointerCapture(e.pointerId)
+		dragFrom = fracAt(e.clientX)
+		dragTo = dragFrom
+	}
+	function onPointerMove(e: PointerEvent) {
+		if (dragFrom == undefined) return
+		dragTo = fracAt(e.clientX)
+	}
+	function onPointerUp(e: PointerEvent) {
+		if (dragFrom == undefined) return
+		const a = dragFrom
+		const b = fracAt(e.clientX)
+		dragFrom = undefined
+		dragTo = undefined
+		// A drag of any meaningful width selects; a click (tiny delta) clears.
+		if (Math.abs(b - a) < 0.012) {
+			onSelect(undefined)
+			return
+		}
+		const lo = Math.min(a, b)
+		const hi = Math.max(a, b)
+		onSelect({ from: Math.floor(fracToMs(lo)), to: Math.ceil(fracToMs(hi)) })
+	}
+
+	// Live drag band (while dragging) or the committed selection.
+	let band = $derived.by(() => {
+		if (dragFrom != undefined && dragTo != undefined) {
+			const lo = Math.min(dragFrom, dragTo)
+			const hi = Math.max(dragFrom, dragTo)
+			return { left: lo, width: hi - lo }
+		}
+		return selFrac
+	})
+
+	function bucketRange(i: number): { from: number; to: number } {
+		const span = Math.max(1, to - from)
+		return { from: from + (i / BUCKETS) * span, to: from + ((i + 1) / BUCKETS) * span }
+	}
+	function fmtTime(ms: number): string {
+		return new Date(ms).toLocaleString(undefined, {
+			month: 'short',
+			day: 'numeric',
+			hour: '2-digit',
+			minute: '2-digit'
+		})
+	}
+</script>
+
+<div class="px-2 pt-1.5 pb-1 select-none">
+	<div
+		bind:this={barsEl}
+		class="relative flex items-end gap-px h-9 cursor-crosshair"
+		role="presentation"
+		onpointerdown={onPointerDown}
+		onpointermove={onPointerMove}
+		onpointerup={onPointerUp}
+		title="Drag to filter to a time range; click to clear"
+	>
+		{#each bins as b, i (i)}
+			{@const total = b.success + b.failure}
+			{@const br = bucketRange(i)}
+			<div
+				class="flex-1 min-w-0 flex flex-col justify-end h-full"
+				style="opacity: {total ? 1 : 0.35}"
+				title={`${fmtTime(br.from)} – ${fmtTime(br.to)}${total ? ` · ${b.success} ok${b.failure ? `, ${b.failure} failed` : ''}` : ' · no runs'}`}
+			>
+				{#if b.failure > 0}
+					<div
+						class="w-full bg-red-400 dark:bg-red-500/80 rounded-t-[1px]"
+						style="height: {(b.failure / maxCount) * 100}%"
+					></div>
+				{/if}
+				{#if b.success > 0}
+					<div
+						class="w-full bg-emerald-400 dark:bg-emerald-500/80"
+						class:rounded-t-[1px]={b.failure === 0}
+						style="height: {(b.success / maxCount) * 100}%"
+					></div>
+				{/if}
+				{#if total === 0}
+					<div class="w-full bg-surface-secondary" style="height: 2px"></div>
+				{/if}
+			</div>
+		{/each}
+		{#if band}
+			<div
+				class="absolute top-0 bottom-0 bg-blue-500/15 border-x border-blue-500/50 pointer-events-none"
+				style="left: {band.left * 100}%; width: {band.width * 100}%"
+			></div>
+		{/if}
+	</div>
+	<!-- Always-on window axis: the period the bars span. -->
+	<div class="flex justify-between text-3xs text-tertiary tabular-nums mt-0.5">
+		<span>{fmtTime(from)}</span>
+		<span>now</span>
+	</div>
+</div>
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
index a46c73f3c1..cd73221626 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
@@ -125,6 +125,13 @@
 		// form (with the auto-generated S3 picker) for the given script.
 		// data_upload has no trigger row; it's a UI-first entry point.
 		onOpenDataUpload?: (scriptPath: string) => void
+		// Script paths the cursor is over in the Activity panel — a thin neutral
+		// ring each. One entry for a single run row; the whole cascade's runs
+		// when hovering a group header. Distinct from `selectedRunPaths`.
+		hoveredPaths?: string[]
+		// Script paths of the expanded (pinned) run(s) — a soft blue ring that
+		// persists until collapsed.
+		selectedRunPaths?: string[]
 	}
 	let {
 		graph,
@@ -143,9 +150,18 @@
 		onEditTrigger,
 		onDeleteTrigger,
 		onOpenWebhook,
-		onOpenDataUpload
+		onOpenDataUpload,
+		hoveredPaths,
+		selectedRunPaths
 	}: Props = $props()
 
+	// `${kind}:${path}` ids for the hovered / pinned runs (both script and flow
+	// variants, since the run row's kind isn't known here).
+	const runIds = (paths: string[] | undefined): Set<string> =>
+		new Set((paths ?? []).flatMap((p) => [`script:${p}`, `flow:${p}`]))
+	let hoveredRunIdSet = $derived(runIds(hoveredPaths))
+	let selectedRunIdSet = $derived(runIds(selectedRunPaths))
+
 	const ADD_NODE_ID = '__add__'
 
 	type BuiltEdge = {
@@ -532,11 +548,25 @@
 			// Compensate for the + node being narrower than its layout slot
 			// so it visually centers over the node(s) below.
 			const xShift = n.id === ADD_NODE_ID ? (NODE.width - ADD_NODE_WIDTH) / 2 : 0
+			// Activity-panel emphasis (purely visual rings, kept off `selected`
+			// which swaps the details pane). Hover wins over pin so the cursor
+			// always tracks.
+			const runClass = hoveredRunIdSet.has(n.id)
+				? 'wm-run-hover'
+				: selectedRunIdSet.has(n.id)
+					? 'wm-run-selected'
+					: undefined
+			// Asset border matching the connecting edge's hue (only when no run
+			// class applies — a node is either a runnable or an asset).
+			const assetEmph = assetEmphasis.get(n.id)
+			const assetClass =
+				assetEmph === 'output' ? 'wm-asset-output' : assetEmph === 'input' ? 'wm-asset-input' : undefined
 			return {
 				id: n.id,
 				type: n.type,
 				position: { x: p.x + xCenter + xShift, y: p.y + 40 },
 				data: n.data,
+				class: runClass ?? assetClass,
 				selected: n.id === selectedId,
 				// All nodes non-draggable: the layout is sugiyama-computed,
 				// dragging would fight the reactive re-layout. Selection is
@@ -562,6 +592,29 @@
 			...(activeRunnableNodeId ? [activeRunnableNodeId] : [])
 		])
 	)
+	// Node ids emphasized from the Activity panel (hovered or pinned runs).
+	// Their incident edges get the same flow animation as a live run, so
+	// hovering a run (or a whole group) traces its data path through the graph.
+	let emphasizedRunIdSet = $derived(new Set<string>([...hoveredRunIdSet, ...selectedRunIdSet]))
+	// Asset nodes adjacent to an emphasized run, tagged by edge family so their
+	// border matches the edge hue: `output` (a write, blue) / `input` (a read or
+	// asset-trigger, gray). Output wins if an asset is both.
+	let assetEmphasis = $derived.by<Map<string, 'input' | 'output'>>(() => {
+		const m = new Map<string, 'input' | 'output'>()
+		if (emphasizedRunIdSet.size === 0) return m
+		for (const e of model.edges) {
+			if (e.kind === 'lineage-write' && emphasizedRunIdSet.has(e.source)) {
+				m.set(e.target, 'output')
+			} else if (
+				(e.kind === 'lineage-read' || e.kind === 'trigger-asset') &&
+				emphasizedRunIdSet.has(e.target) &&
+				m.get(e.source) !== 'output'
+			) {
+				m.set(e.source, 'input')
+			}
+		}
+		return m
+	})
 	let flowEdges = $derived.by(() =>
 		model.edges
 			// Anchor edges are layout-only.
@@ -570,6 +623,12 @@
 				const touchesActiveRunnable =
 					activeRunnableIdSet.size > 0 &&
 					(activeRunnableIdSet.has(e.source) || activeRunnableIdSet.has(e.target))
+				// Edge incident to an Activity-panel hovered/pinned run — animate
+				// it too so the row↔graph link is unmistakable.
+				const touchesEmphasis =
+					emphasizedRunIdSet.size > 0 &&
+					(emphasizedRunIdSet.has(e.source) || emphasizedRunIdSet.has(e.target))
+				const flowAnimated = touchesActiveRunnable || touchesEmphasis
 				let style: string
 				let animated = false
 				let markerColor: string | undefined = undefined
@@ -586,16 +645,16 @@
 				switch (e.kind) {
 					case 'lineage-write':
 						style = 'stroke: rgb(59 130 246); stroke-width: 2px;'
-						animated = touchesActiveRunnable
+						animated = flowAnimated
 						markerColor = 'rgb(59 130 246)'
 						break
 					case 'lineage-read':
 						style = 'stroke: rgb(156 163 175); stroke-width: 1.25px;'
-						animated = touchesActiveRunnable
+						animated = flowAnimated
 						break
 					case 'trigger-asset':
 						style = 'stroke: rgb(107 114 128); stroke-width: 2px;'
-						animated = touchesActiveRunnable
+						animated = flowAnimated
 						markerColor = 'rgb(107 114 128)'
 						label = 'triggers'
 						labelStyle = 'fill: rgb(107 114 128); font-size: 10px; font-weight: 600;'
@@ -743,4 +802,21 @@
 	:global(.svelte-flow__node.selected .drop-shadow-sm) {
 		@apply outline outline-2 outline-blue-500;
 	}
+	/* Activity-panel emphasis — soft, monochromatic, less prominent than the
+	   blue details selection above. Hover is a thin neutral ring (transient);
+	   pinning an expanded run is a soft-blue ring. */
+	:global(.svelte-flow__node.wm-run-hover .drop-shadow-sm) {
+		@apply outline outline-1 outline-gray-400 dark:outline-gray-500;
+	}
+	:global(.svelte-flow__node.wm-run-selected .drop-shadow-sm) {
+		@apply outline outline-2 outline-blue-400/70;
+	}
+	/* Assets adjacent to an emphasized run — border hue matches the edge:
+	   blue for a write (output), gray for a read / asset-trigger (input). */
+	:global(.svelte-flow__node.wm-asset-output .drop-shadow-sm) {
+		@apply outline outline-2 outline-blue-400/70;
+	}
+	:global(.svelte-flow__node.wm-asset-input .drop-shadow-sm) {
+		@apply outline outline-2 outline-gray-400;
+	}
 </style>
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
index df04960f94..71d4f85921 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
@@ -71,6 +71,12 @@
 		// edited past the seeded template. Fires on every keystroke that
 		// changes the inferred set.
 		onAssetsChange?: (scriptPath: string | undefined, assets: AssetWithAltAccessType[]) => void
+		// Emits the live editor buffer on every keystroke so the parent can
+		// autosave the in-flight content WITHOUT waiting for the pane teardown
+		// (`onDraftPersist`). `onDraftPersist` stays the authoritative commit on
+		// navigate-away; this is the continuous feed that keeps the draft's
+		// server copy current while the user is still typing.
+		onContentChange?: (scriptPath: string | undefined, content: string) => void
 		// Fires when the user navigates away from a draft (selects another
 		// node, closes the pane, etc.) — the parent must persist the new
 		// content + write outputs into its drafts Map. Without this,
@@ -196,6 +202,7 @@
 		onDiscard,
 		onAnnotationsChange,
 		onAssetsChange,
+		onContentChange,
 		onDraftPersist,
 		onScriptRenamed,
 		onScriptRemoved,
@@ -539,6 +546,10 @@
 		if (readOnly) return
 		onAssetsChange?.(script?.path, liveBodyAssets ?? [])
 	})
+	$effect(() => {
+		if (readOnly) return
+		onContentChange?.(script?.path, script?.content ?? '')
+	})
 
 	async function save() {
 		if (!script) return
@@ -552,7 +563,7 @@
 			} catch {
 				sendUserToast(`Could not parse code, are you sure it is valid?`, true)
 			}
-			await ScriptService.createScript({
+			const newHash = await ScriptService.createScript({
 				workspace,
 				requestBody: {
 					...script,
@@ -579,6 +590,11 @@
 					assets: (liveBodyAssets ?? []) as any
 				}
 			})
+			// Chain the NEXT save off the version we just created. Without this
+			// a second save re-sends the now-stale `parent_hash`, which the
+			// backend rejects as a lineage fork ("no 2 scripts can have the
+			// same parent").
+			if (typeof newHash === 'string' && newHash) script.hash = newHash
 			sendUserToast(`Saved ${script.path}`)
 			if (isDraft) {
 				onDraftSaved?.(script.path)
@@ -587,12 +603,56 @@
 				onPersistedSaved?.(script.path)
 			}
 		} catch (e: any) {
-			sendUserToast(`Save failed: ${e?.body ?? e?.message ?? e}`, true)
+			const msg = String(e?.body ?? e?.message ?? e)
+			// The deployed head moved off our `parent_hash` since we opened it —
+			// someone else (or another tab) deployed in between. Offer an
+			// explicit resolution instead of a cryptic lineage error.
+			if (/lineage must be linear|same parent_hash/i.test(msg)) {
+				conflictOpen = true
+			} else {
+				sendUserToast(`Save failed: ${msg}`, true)
+			}
 		} finally {
 			saving = false
 		}
 	}
 
+	// --- Deploy conflict resolution (another deploy landed since we opened) ---
+	let conflictOpen = $state(false)
+	let resolvingConflict = $state(false)
+	async function overwriteWithMine() {
+		if (!script) return
+		resolvingConflict = true
+		try {
+			// Re-base onto the current head so the retry is a linear deploy on
+			// top of theirs (our content wins).
+			const latest = await ScriptService.getScriptByPath({ workspace, path: script.path })
+			script.hash = String(latest.hash)
+			conflictOpen = false
+			await save()
+		} catch (e: any) {
+			sendUserToast(`Could not resolve conflict: ${e?.body ?? e?.message ?? e}`, true)
+		} finally {
+			resolvingConflict = false
+		}
+	}
+	async function viewLatestVersion() {
+		if (!script) return
+		resolvingConflict = true
+		try {
+			const latest = await ScriptService.getScriptByPath({ workspace, path: script.path })
+			// Replace the editor buffer with the latest deployed version. The
+			// user's unsaved edits are dropped — surfaced clearly in the modal.
+			script = structuredClone($state.snapshot(latest) as Script)
+			conflictOpen = false
+			sendUserToast('Loaded the latest deployed version')
+		} catch (e: any) {
+			sendUserToast(`Could not load the latest version: ${e?.body ?? e?.message ?? e}`, true)
+		} finally {
+			resolvingConflict = false
+		}
+	}
+
 	let isScriptView = $derived(
 		isDraft || (selection?.kind === 'runnable' && selection.runnable_kind === 'script')
 	)
@@ -1121,6 +1181,72 @@
 	</div>
 {/if}
 
+{#if conflictOpen}
+	<!-- Deploy conflict: the path's deployed head moved off our parent_hash
+	     since we opened it (another user / tab deployed in between). -->
+	<div
+		transition:fade={{ duration: 100 }}
+		class="fixed top-0 bottom-0 left-0 right-0 z-[9999]"
+		role="dialog"
+	>
+		<div class="fixed inset-0 bg-gray-500 bg-opacity-75"></div>
+		<div class="fixed inset-0 z-10 overflow-y-auto">
+			<div class="flex min-h-full items-center justify-center p-4">
+				<div
+					class="relative transform overflow-hidden rounded-lg bg-surface px-4 pt-5 pb-4 text-left shadow-xl sm:my-8 sm:w-full sm:max-w-lg sm:p-6"
+				>
+					<div class="flex">
+						<div
+							class="flex h-12 w-12 items-center justify-center rounded-full bg-amber-100 dark:bg-amber-800/50"
+						>
+							<AlertTriangle class="text-amber-500 dark:text-amber-400" />
+						</div>
+						<div class="ml-4 flex-1">
+							<h3 class="text-lg font-medium text-primary">This script changed since you opened it</h3>
+							<div class="mt-2 text-sm text-secondary flex flex-col gap-2">
+								<p>
+									<span class="font-mono text-xs">{script?.path ?? ''}</span> was deployed by someone else
+									(or another tab) while you were editing, so saving on top of your version would fork its
+									lineage.
+								</p>
+								<p>
+									<span class="font-medium">Keep my version</span> deploys your changes on top of the latest.
+									<span class="font-medium">View latest</span> loads the newly-deployed version, discarding
+									your unsaved edits.
+								</p>
+							</div>
+						</div>
+					</div>
+					<div class="flex items-center gap-2 flex-row-reverse mt-4">
+						<Button disabled={resolvingConflict} onclick={overwriteWithMine} variant="accent" size="sm">
+							{#if resolvingConflict}
+								<Loader2 class="animate-spin" />
+							{/if}
+							<span class="min-w-20">Keep my version</span>
+						</Button>
+						<Button
+							disabled={resolvingConflict}
+							onclick={() => (conflictOpen = false)}
+							variant="default"
+							size="sm"
+						>
+							Cancel
+						</Button>
+						<Button
+							disabled={resolvingConflict}
+							onclick={viewLatestVersion}
+							variant="contained"
+							size="sm"
+						>
+							View latest version
+						</Button>
+					</div>
+				</div>
+			</div>
+		</div>
+	</div>
+{/if}
+
 <style>
 	/* Border-color pulse for the throwaway overlay appended to <body> when the
 	   user clicks a data_upload source node (hence :global). border-box keeps
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineActivityPanel.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineActivityPanel.svelte
index c5da8d79af..90e531e858 100644
--- a/frontend/src/lib/components/assets/AssetGraph/PipelineActivityPanel.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/PipelineActivityPanel.svelte
@@ -20,27 +20,223 @@
 	} from 'lucide-svelte'
 	import { twMerge } from 'tailwind-merge'
 	import { untrack } from 'svelte'
-	import { isActiveEvent, type PipelineEvent } from './activeRunnables.svelte'
+	import { isActiveEvent, type EventStatus, type PipelineEvent } from './activeRunnables.svelte'
+	import type { DispatchEdge } from './pipelineHistory.svelte'
+	import { RotateCcw, Workflow } from 'lucide-svelte'
+	import ActivityHistogram from './ActivityHistogram.svelte'
 
 	interface Props {
 		// Merged history + live events, newest-first (page owns the merge —
 		// the live poll wins on id collisions since it carries fresher status).
 		events: PipelineEvent[]
+		// Asset-cascade producer→child edges (by job id) used to group the
+		// runs of one cascade together.
+		edges?: DispatchEdge[]
 		loading?: boolean
 		// History preload hit its page cap before the days cutoff.
 		truncated?: boolean
 		error?: string | undefined
 		days: number
 		onDaysChange: (days: number) => void
+		// Hover a run row → emphasize its node(s) on the canvas; a group header
+		// passes the whole cascade's paths. `undefined` clears.
+		onHoverRun?: (paths: string[] | undefined) => void
+		// Expand/collapse a run → sticky node emphasis (undefined clears).
+		onSelectRun?: (paths: string[] | undefined) => void
 	}
 
-	let { events, loading = false, truncated = false, error, days, onDaysChange }: Props = $props()
+	let {
+		events,
+		edges = [],
+		loading = false,
+		truncated = false,
+		error,
+		days,
+		onDaysChange,
+		onHoverRun,
+		onSelectRun
+	}: Props = $props()
 
+	type CascadeGroup = {
+		// Component representative id (stable per cascade).
+		key: string
+		// Earliest originating (non-asset) run — the group header. Undefined when
+		// the originating run fell outside the loaded window.
+		root: PipelineEvent | undefined
+		// Additional originating triggers beyond `root` (joins fed by >1 source).
+		extraTriggers: number
+		// All loaded runs of this cascade: roots first, then newest-first.
+		members: PipelineEvent[]
+		// Worth grouping (more than one run, or an edge reaches outside window).
+		isCascade: boolean
+		// Aggregate status across members (running > failure > success).
+		status: EventStatus
+		// Newest member time — drives group ordering.
+		latestAt: string
+	}
+
+	// Histogram window (last `days`) and the brushed sub-range. The histogram
+	// always shows the full window; the list + groups filter to the brush.
+	let windowBounds = $derived.by(() => {
+		void events // refresh as new runs arrive
+		const to = Date.now()
+		return { from: to - days * 86400000, to }
+	})
+	let selectedRange = $state<{ from: number; to: number } | undefined>(undefined)
+	// Drop a stale brush when the day window changes under it.
+	$effect(() => {
+		void days
+		untrack(() => (selectedRange = undefined))
+	})
+	let filteredEvents = $derived.by(() => {
+		const r = selectedRange
+		if (!r) return events
+		return events.filter((e) => {
+			const t = Date.parse(e.at)
+			return !isNaN(t) && t >= r.from && t <= r.to
+		})
+	})
+	// Quick reset: drop the brush and return to the default 30-day window.
+	function resetWindow() {
+		selectedRange = undefined
+		if (days !== 30) onDaysChange(30)
+	}
+	function fmtRange(r: { from: number; to: number }): string {
+		const opt: Intl.DateTimeFormatOptions = {
+			month: 'short',
+			day: 'numeric',
+			hour: '2-digit',
+			minute: '2-digit'
+		}
+		return `${new Date(r.from).toLocaleString(undefined, opt)} – ${new Date(r.to).toLocaleString(undefined, opt)}`
+	}
+
+	// Group cascades by the CONNECTED COMPONENT of the dispatch graph, so an
+	// AND-join (fed by several trigger chains) folds into one group rather than
+	// fragmenting by whichever input completed it last. Edges come by job id
+	// from `dispatched` rows; `join_pending` inputs (no child yet) are linked to
+	// the next firing of the same subscriber so the pre-completion contributors
+	// join the component too.
+	let groups = $derived.by<CascadeGroup[]>(() => {
+		// Producer→child edges (job id). `childSet` = jobs with an incoming edge
+		// (i.e. not an originating run).
+		const childEdges: Array<[string, string]> = []
+		const childSet = new Set<string>()
+		const inEdges = new Set<string>()
+		// Firings per subscriber path (a `dispatched` row), time-sorted, to map
+		// each join_pending input forward to the child it contributed to.
+		const firingsBySub = new Map<string, Array<{ child: string; at: number }>>()
+		for (const e of edges) {
+			if (e.outcome === 'dispatched' && e.child_job_id) {
+				childEdges.push([e.producer_job_id, e.child_job_id])
+				childSet.add(e.child_job_id)
+				const arr = firingsBySub.get(e.subscriber_path)
+				const f = { child: e.child_job_id, at: Date.parse(e.created_at) }
+				if (arr) arr.push(f)
+				else firingsBySub.set(e.subscriber_path, [f])
+			}
+		}
+		for (const arr of firingsBySub.values()) arr.sort((a, b) => a.at - b.at)
+		for (const e of edges) {
+			if (e.outcome !== 'join_pending') continue
+			const firings = firingsBySub.get(e.subscriber_path)
+			if (!firings) continue
+			const t = Date.parse(e.created_at)
+			const firing = firings.find((f) => f.at >= t)
+			if (firing) {
+				childEdges.push([e.producer_job_id, firing.child])
+				childSet.add(firing.child)
+			}
+		}
+		for (const [p, c] of childEdges) {
+			inEdges.add(p)
+			inEdges.add(c)
+		}
+		// Union-find over the edges.
+		const parent = new Map<string, string>()
+		const find = (x: string): string => {
+			let r = x
+			while (parent.has(r) && parent.get(r) !== r) r = parent.get(r)!
+			return r
+		}
+		const union = (a: string, b: string) => {
+			if (!parent.has(a)) parent.set(a, a)
+			if (!parent.has(b)) parent.set(b, b)
+			const ra = find(a)
+			const rb = find(b)
+			if (ra !== rb) parent.set(ra, rb)
+		}
+		for (const [p, c] of childEdges) union(p, c)
+
+		const byComponent = new Map<string, PipelineEvent[]>()
+		for (const e of filteredEvents) {
+			const k = parent.has(e.id) ? find(e.id) : e.id
+			const arr = byComponent.get(k)
+			if (arr) arr.push(e)
+			else byComponent.set(k, [e])
+		}
+		const out: CascadeGroup[] = []
+		for (const [key, members] of byComponent) {
+			const roots = members
+				.filter((m) => !childSet.has(m.id))
+				.sort((a, b) => (a.at < b.at ? -1 : a.at > b.at ? 1 : 0))
+			members.sort((a, b) => {
+				const aRoot = !childSet.has(a.id)
+				const bRoot = !childSet.has(b.id)
+				if (aRoot !== bRoot) return aRoot ? -1 : 1
+				if (aRoot) return a.at < b.at ? -1 : a.at > b.at ? 1 : 0 // roots earliest-first
+				return a.at < b.at ? 1 : a.at > b.at ? -1 : 0 // hops newest-first
+			})
+			const isCascade = members.length > 1 || members.some((m) => inEdges.has(m.id))
+			const anyActive = members.some((m) => m.status === 'running' || m.status === 'queued')
+			const anyFail = members.some((m) => m.status === 'failure')
+			const status: EventStatus = anyActive ? 'running' : anyFail ? 'failure' : 'success'
+			const latestAt = members.reduce((mx, m) => (m.at > mx ? m.at : mx), members[0].at)
+			out.push({
+				key,
+				root: roots[0],
+				extraTriggers: Math.max(0, roots.length - 1),
+				members,
+				isCascade,
+				status,
+				latestAt
+			})
+		}
+		out.sort((a, b) => (a.latestAt < b.latestAt ? 1 : a.latestAt > b.latestAt ? -1 : 0))
+		return out
+	})
+
+	// Collapsed cascade groups (members hidden). Default expanded.
+	let collapsedGroups = $state<Set<string>>(new Set())
+	function toggleGroup(key: string) {
+		const next = new Set(collapsedGroups)
+		if (next.has(key)) next.delete(key)
+		else next.add(key)
+		collapsedGroups = next
+	}
+
+	// The cascade's trigger label, shown on the group header — the earliest
+	// originating run's source. A loaded root is never asset-triggered (an asset
+	// hop always has a producer), so its `source` is the real origin; an
+	// unloaded root means the originating run fell outside the window.
+	function groupTrigger(g: CascadeGroup): string {
+		if (!g.root) return 'cascade'
+		return g.root.source === 'schedule' ? 'schedule' : 'run'
+	}
+	function groupTitle(g: CascadeGroup): string {
+		return g.root?.path ?? 'cascade (upstream outside window)'
+	}
+
+	// Window options as fractions of a day (the fetch cutoff is `days * 1d`).
 	const DAY_OPTIONS = [
+		{ label: 'Last hour', value: 1 / 24 },
+		{ label: 'Last 24 hours', value: 1 },
+		{ label: 'Last 48 hours', value: 2 },
 		{ label: 'Last 7 days', value: 7 },
 		{ label: 'Last 30 days', value: 30 },
 		{ label: 'Last 90 days', value: 90 }
 	]
+	let windowLabel = $derived(DAY_OPTIONS.find((o) => o.value === days)?.label ?? `Last ${days} days`)
 
 	// Excludes future-scheduled queued jobs (a schedule's next planned run
 	// is not activity) — see isActiveEvent.
@@ -73,8 +269,11 @@
 		})
 	})
 
-	function toggle(id: string) {
-		expandedId = expandedId === id ? undefined : id
+	function toggle(id: string, path: string) {
+		const willExpand = expandedId !== id
+		expandedId = willExpand ? id : undefined
+		// Expanding a run pins its node emphasis; collapsing clears it.
+		onSelectRun?.(willExpand ? [path] : undefined)
 	}
 
 	function isRunningJob(j: Job): boolean {
@@ -152,6 +351,33 @@
 		</div>
 	</div>
 
+	{#if events.length > 0}
+		<div class="border-b shrink-0">
+			<ActivityHistogram
+				{events}
+				from={windowBounds.from}
+				to={windowBounds.to}
+				selected={selectedRange}
+				onSelect={(r) => (selectedRange = r)}
+			/>
+			{#if selectedRange || days !== 30}
+				<div class="flex items-center justify-between px-2 pb-1.5 text-2xs text-tertiary">
+					<span class="truncate">
+						{selectedRange ? fmtRange(selectedRange) : windowLabel}
+					</span>
+					<button
+						type="button"
+						class="flex items-center gap-1 px-1.5 py-0.5 rounded-sm hover:bg-surface-hover text-secondary shrink-0"
+						onclick={resetWindow}
+						title="Reset to the last 30 days"
+					>
+						<RotateCcw size={11} /> Reset
+					</button>
+				</div>
+			{/if}
+		</div>
+	{/if}
+
 	<div class="flex-1 min-h-0 overflow-y-auto px-1 py-1">
 		{#if error}
 			<div class="px-3 py-4 text-xs text-red-600 dark:text-red-400">
@@ -164,17 +390,21 @@
 				</div>
 			{:else}
 				<div class="px-3 py-6 text-center text-xs text-tertiary">
-					No runs in the last {days} days — executions of this pipeline will appear here live.
+					No runs in this window ({windowLabel.toLowerCase()}) — executions of this pipeline will
+					appear here live.
 				</div>
 			{/if}
 		{:else}
-			{#each events as e (e.id)}
+			{#snippet runRow(e: PipelineEvent, isMember: boolean, showSource: boolean)}
 				{@const expanded = expandedId === e.id}
 				<button
 					type="button"
-					onclick={() => toggle(e.id)}
+					onclick={() => toggle(e.id, e.path)}
+					onmouseenter={() => onHoverRun?.([e.path])}
+					onmouseleave={() => onHoverRun?.(undefined)}
 					class={twMerge(
 						'w-full flex items-center gap-2 px-2 py-1 rounded-sm hover:bg-surface-hover text-xs text-left',
+						isMember && 'pl-7',
 						expanded && 'bg-surface-selected'
 					)}
 					title={expanded ? 'Collapse run details' : 'Expand run details'}
@@ -205,16 +435,18 @@
 					<span class="flex-1 min-w-0 truncate font-mono text-2xs" title={e.path}>
 						{e.path}
 					</span>
-					<span
-						class={twMerge(
-							'shrink-0 px-1 py-0.5 rounded-sm text-3xs leading-none',
-							e.source === 'schedule'
-								? 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300'
-								: 'bg-gray-100 dark:bg-gray-800 text-gray-600 dark:text-gray-400'
-						)}
-					>
-						{e.source}
-					</span>
+					{#if showSource}
+						<span
+							class={twMerge(
+								'shrink-0 px-1 py-0.5 rounded-sm text-3xs leading-none',
+								e.source === 'schedule'
+									? 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300'
+									: 'bg-gray-100 dark:bg-gray-800 text-gray-600 dark:text-gray-400'
+							)}
+						>
+							{e.source}
+						</span>
+					{/if}
 					<span class="shrink-0 text-3xs text-tertiary tabular-nums w-14 text-right">
 						{isFutureScheduled(e) && e.scheduledFor ? inTxt(e.scheduledFor) : ago(e.at)}
 					</span>
@@ -271,6 +503,71 @@
 						{/if}
 					</div>
 				{/if}
+			{/snippet}
+
+			{#if groups.length === 0}
+				<div class="px-3 py-6 text-center text-xs text-tertiary">
+					No runs in the selected time range.
+				</div>
+			{/if}
+			{#each groups as g (g.key)}
+				{#if g.isCascade}
+					{@const collapsed = collapsedGroups.has(g.key)}
+					<!-- Cascade group header: the originating run + its trigger, the
+					     aggregate status, and a member count. Collapses the hops. -->
+					<button
+						type="button"
+						onclick={() => toggleGroup(g.key)}
+						onmouseenter={() => onHoverRun?.(g.members.map((m) => m.path))}
+						onmouseleave={() => onHoverRun?.(undefined)}
+						class="w-full flex items-center gap-2 px-2 py-1 rounded-sm hover:bg-surface-hover text-xs text-left"
+						title={collapsed ? 'Expand cascade runs' : 'Collapse cascade runs'}
+					>
+						<span class="shrink-0 text-tertiary">
+							{#if collapsed}<ChevronRight size={12} />{:else}<ChevronDown size={12} />{/if}
+						</span>
+						<span class="shrink-0">
+							{#if g.status === 'running'}
+								<Loader2 size={12} class="animate-spin text-blue-600 dark:text-blue-400" />
+							{:else if g.status === 'failure'}
+								<XCircle size={12} class="text-red-600 dark:text-red-400" />
+							{:else}
+								<CheckCircle2 size={12} class="text-emerald-600 dark:text-emerald-400" />
+							{/if}
+						</span>
+						<Workflow size={12} class="shrink-0 text-indigo-500" />
+						<span class="flex-1 min-w-0 truncate font-mono text-2xs" title={groupTitle(g)}>
+							{groupTitle(g)}
+						</span>
+						<span
+							class={twMerge(
+								'shrink-0 px-1 py-0.5 rounded-sm text-3xs leading-none',
+								groupTrigger(g) === 'schedule'
+									? 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300'
+									: 'bg-indigo-100 dark:bg-indigo-900/40 text-indigo-700 dark:text-indigo-300'
+							)}
+						>
+							{groupTrigger(g)}
+						</span>
+						{#if g.extraTriggers > 0}
+							<span
+								class="shrink-0 px-1 py-0.5 rounded-sm text-3xs leading-none bg-gray-100 dark:bg-gray-800 text-gray-600 dark:text-gray-400"
+								title={`Join fed by ${g.extraTriggers + 1} triggers`}>+{g.extraTriggers}</span
+							>
+						{/if}
+						<span class="shrink-0 text-3xs text-tertiary tabular-nums">{g.members.length} runs</span>
+						<span class="shrink-0 text-3xs text-tertiary tabular-nums w-14 text-right">
+							{ago(g.latestAt)}
+						</span>
+					</button>
+					{#if !collapsed}
+						{#each g.members as m (m.id)}
+							{@render runRow(m, true, false)}
+						{/each}
+					{/if}
+				{:else}
+					{@render runRow(g.members[0], false, true)}
+				{/if}
 			{/each}
 			{#if truncated}
 				<div class="px-3 py-2 text-center text-2xs text-tertiary">
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts
index 7a5ab134d3..bfa316979e 100644
--- a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts
@@ -1,6 +1,21 @@
 import { JobService } from '$lib/gen'
 import type { PipelineEvent } from './activeRunnables.svelte'
 
+// One asset-cascade dispatch record (from the dispatch_event table). A
+// `dispatched` row carries the resolved `child_job_id` (a producer→child job
+// edge); a `join_pending` row is a pre-completion AND-join input with no child
+// yet — the panel links it to the eventual child of the same subscriber so a
+// join's separate trigger chains land in one group.
+export type DispatchEdge = {
+	producer_job_id: string
+	child_job_id?: string
+	subscriber_path: string
+	outcome: 'dispatched' | 'join_pending'
+	asset_path: string
+	asset_kind: string
+	created_at: string
+}
+
 // Page size is the API max; 3 pages keeps the preload bounded on noisy
 // folders — the panel notes the truncation instead of silently capping.
 const PER_PAGE = 100
@@ -21,6 +36,7 @@ export function usePipelineHistory(
 	getEnabled: () => boolean
 ) {
 	let events = $state<PipelineEvent[]>([])
+	let edges = $state<DispatchEdge[]>([])
 	let loading = $state(false)
 	let truncated = $state(false)
 	let error = $state<string | undefined>(undefined)
@@ -61,7 +77,21 @@ export function usePipelineHistory(
 				}
 				sawFullPage = rows.length === PER_PAGE
 			}
+			// Cascade edges for the same window. Best-effort: a failure here
+			// must not blank the activity list, so it only degrades grouping.
+			let edgeRows: DispatchEdge[] = []
+			try {
+				edgeRows = (await JobService.listAssetDispatchEdges({
+					workspace: ws,
+					pathStart: prefix,
+					createdAfter: cutoff
+				})) as DispatchEdge[]
+			} catch (e) {
+				console.warn('failed to load pipeline dispatch edges', e)
+			}
+			if (gen !== myGen) return
 			events = out
+			edges = edgeRows
 			truncated = sawFullPage
 		} catch (e: any) {
 			if (gen !== myGen) return
@@ -89,6 +119,10 @@ export function usePipelineHistory(
 		get events() {
 			return events
 		},
+		/** Asset-cascade producer→child edges over the same window. */
+		get edges() {
+			return edges
+		},
 		get loading() {
 			return loading
 		},
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
index fb0932cf35..8025216921 100644
--- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
@@ -5,8 +5,15 @@
 	import Button from '$lib/components/common/button/Button.svelte'
 	import DropdownV2 from '$lib/components/DropdownV2.svelte'
 	import AssetGraphCanvas from '$lib/components/assets/AssetGraph/AssetGraphCanvas.svelte'
-	import { useActiveRunnableIds } from '$lib/components/assets/AssetGraph/activeRunnables.svelte'
-	import type { PipelineEvent } from '$lib/components/assets/AssetGraph/activeRunnables.svelte'
+	import {
+		useActiveRunnableIds,
+		isActiveEvent
+	} from '$lib/components/assets/AssetGraph/activeRunnables.svelte'
+	import type {
+		PipelineEvent,
+		RunnableRunState,
+		RunStatus
+	} from '$lib/components/assets/AssetGraph/activeRunnables.svelte'
 	import { usePipelineHistory } from '$lib/components/assets/AssetGraph/pipelineHistory.svelte'
 	import PipelineEventLog from '$lib/components/assets/AssetGraph/PipelineEventLog.svelte'
 	import PipelineActivityPanel from '$lib/components/assets/AssetGraph/PipelineActivityPanel.svelte'
@@ -47,6 +54,7 @@
 	import { decodeState, encodeState } from '$lib/utils'
 	import { DraftService } from '$lib/gen'
 	import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
+	import AutosaveIndicator from '$lib/components/AutosaveIndicator.svelte'
 	import { onMount, tick, untrack } from 'svelte'
 	import { aiChatManager } from '$lib/components/copilot/chat/AIChatManager.svelte'
 	import {
@@ -113,6 +121,18 @@
 	let urlMode = $derived(page.url.searchParams.get('mode'))
 	let mode = $derived<PipelineMode>(isOperator ? 'view' : urlMode === 'edit' ? 'edit' : 'view')
 	function setMode(m: PipelineMode, opts?: { replace?: boolean }) {
+		// Switching edit→view re-surfaces the activity feed: drop the edit-mode
+		// selection / open draft / hidden-pane state so view opens on Activity
+		// rather than a stale details pane (mirrors toggleActivity's show path).
+		if (m === 'view' && mode === 'edit') {
+			selection = undefined
+			activeDraftPath = undefined
+			panelHidden = false
+			liveAnnotations = {
+				scriptPath: undefined,
+				annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
+			}
+		}
 		const url = new URL(page.url)
 		if (m === 'view') url.searchParams.delete('mode')
 		else url.searchParams.set('mode', m)
@@ -269,6 +289,9 @@
 	// just-loaded value itself) isn't re-POSTed.
 	let draftsHydrated = $state(false)
 	let lastPersistedBundle: string | undefined = undefined
+	// True once a bundle was restored from the DB on load — drives the
+	// AutosaveIndicator's one-shot "Loaded from draft" hint.
+	let loadedFromDbDraft = $state(false)
 
 	function restoreBundle(bundle: PipelineDraftBundle) {
 		if (Array.isArray(bundle.drafts)) {
@@ -329,6 +352,7 @@
 				if (row?.value) {
 					bundle = row.value as PipelineDraftBundle
 					serverSavedAt = row.created_at
+					loadedFromDbDraft = true
 				}
 			}
 			// One-time migration: no DB draft yet, but an older build left a
@@ -377,14 +401,29 @@
 				? extractWrites(liveBodyAssets.assets)
 				: undefined
 		const liveWritesPath = liveBodyAssets.scriptPath
+		// Live editor buffer for the open draft. `onDraftPersist` only commits
+		// content into the Map on pane teardown, so without this overlay the
+		// autosave (and a crash reload) would lag a whole editing session behind.
+		const liveContentPath =
+			liveContent.scriptPath != undefined && drafts.has(liveContent.scriptPath)
+				? liveContent.scriptPath
+				: undefined
+		const liveContentValue = liveContent.content
 		const serialized = Array.from(drafts.entries()).map(([p, d]) => {
-			if (liveWritesSnapshot != undefined && liveWritesPath === p) {
-				return [
-					p,
-					{ ...d, outputAssets: liveWritesSnapshot.length > 0 ? liveWritesSnapshot : undefined }
-				] as [string, Draft]
+			const outputAssets =
+				liveWritesSnapshot != undefined && liveWritesPath === p
+					? liveWritesSnapshot.length > 0
+						? liveWritesSnapshot
+						: undefined
+					: d.outputAssets
+			const script =
+				liveContentPath === p && d.script.content !== liveContentValue
+					? { ...d.script, content: liveContentValue }
+					: d.script
+			if (script === d.script && outputAssets === d.outputAssets) {
+				return [p, d] as [string, Draft]
 			}
-			return [p, d] as [string, Draft]
+			return [p, { ...d, script, outputAssets }] as [string, Draft]
 		})
 		const activePath = activeDraftPath
 		const key = storageKey
@@ -449,6 +488,16 @@
 		assets: AssetWithAltAccessType[]
 	}>({ scriptPath: undefined, assets: [] })
 
+	// The open draft's live editor buffer, emitted by the pane on every
+	// keystroke (`onContentChange`). The persist effect overlays it onto the
+	// Map's copy so autosave reflects in-progress edits — `onDraftPersist`
+	// only commits content into the Map on pane teardown, which would leave
+	// autosave a full editing session behind.
+	let liveContent = $state<{ scriptPath: string | undefined; content: string }>({
+		scriptPath: undefined,
+		content: ''
+	})
+
 	// Only-add cache of (script_path → body content) populated lazily by
 	// `bodyFetchEffect` for every script in the current folder. We never
 	// remove entries: stale keys (renamed-away, deleted) are simply ignored
@@ -865,6 +914,9 @@
 		if (liveBodyAssets.scriptPath === path) {
 			liveBodyAssets = { scriptPath: undefined, assets: [] }
 		}
+		if (liveContent.scriptPath === path) {
+			liveContent = { scriptPath: undefined, content: '' }
+		}
 		clearSaveError(path)
 	}
 
@@ -904,6 +956,9 @@
 		if (liveBodyAssets.scriptPath === oldPath) {
 			liveBodyAssets = { ...liveBodyAssets, scriptPath: newPath }
 		}
+		if (liveContent.scriptPath === oldPath) {
+			liveContent = { ...liveContent, scriptPath: newPath }
+		}
 		// `inferredWritesByPath` / `inferredReadsByPath` /
 		// `annotatedNativeKindsByPath` are derived from `g.runnables` ×
 		// the body caches, so a rename naturally re-derives them when the
@@ -1090,6 +1145,9 @@
 		// next reactive tick.
 		liveBodyAssets = { scriptPath, assets }
 	}
+	function handleContentChange(scriptPath: string | undefined, content: string) {
+		liveContent = { scriptPath, content }
+	}
 	function handleDraftPersist(
 		p: string,
 		snapshot: {
@@ -1412,6 +1470,9 @@
 				}
 				liveBodyAssets = { scriptPath: undefined, assets: [] }
 			}
+			if (liveContent.scriptPath != undefined) {
+				liveContent = { scriptPath: undefined, content: '' }
+			}
 		})
 	})
 	// Drafts are only addressable in view mode while the drafts overlay is
@@ -1477,11 +1538,13 @@
 	// View-mode activity panel: historical runs preloaded for the last N
 	// days (user-configurable, persisted) merged with the live poll's events.
 	const ACTIVITY_DAYS_KEY = 'pipeline-activity-days'
+	// Allowed window sizes in days (fractions for the sub-day ranges).
+	const ACTIVITY_WINDOW_DAYS = [1 / 24, 1, 2, 7, 30, 90]
 	let activityDays = $state(30)
 	onMount(() => {
 		if (typeof localStorage === 'undefined') return
 		const stored = Number(localStorage.getItem(ACTIVITY_DAYS_KEY))
-		if ([7, 30, 90].includes(stored)) activityDays = stored
+		if (ACTIVITY_WINDOW_DAYS.some((d) => Math.abs(d - stored) < 1e-9)) activityDays = stored
 	})
 	function setActivityDays(days: number) {
 		activityDays = days
@@ -1503,6 +1566,33 @@
 		for (const e of activeRunnables.events) byId.set(e.id, e)
 		return Array.from(byId.values()).sort((a, b) => b.at.localeCompare(a.at))
 	})
+	// Node run-count/status badges, derived from the SAME merged event set the
+	// Activity panel shows (historic preload + live poll) so the graph badges
+	// and the panel never disagree. `activityEvents` is newest-first, so the
+	// first event per runnable sets the badge status and the rest add to the
+	// count. Future-scheduled queued runs aren't activity — skip them.
+	let mergedRunStates = $derived.by<Map<string, RunnableRunState>>(() => {
+		const m = new Map<string, RunnableRunState>()
+		for (const e of activityEvents) {
+			if (e.status === 'queued' && !isActiveEvent(e)) continue
+			const id = `${e.kind}:${e.path}`
+			const status: RunStatus =
+				e.status === 'running' || e.status === 'queued'
+					? 'running'
+					: e.status === 'failure'
+						? 'failure'
+						: 'success'
+			const cur = m.get(id)
+			if (cur) cur.runs += 1
+			else m.set(id, { status, runs: 1 })
+		}
+		return m
+	})
+	// Activity-panel → canvas node emphasis. Hover and the pinned expanded run
+	// are fed to the canvas as separate path sets so they render distinctly; a
+	// group-header hover passes the whole cascade.
+	let activityHoverPaths = $state<string[]>([])
+	let activitySelectPaths = $state<string[]>([])
 	// Release the zero-latency `activeRunnable` hint for a per-node run of a
 	// script that isn't open in the pane: such runs have no editor Test callback
 	// to clear it, so without this the node's badge stays 'running' forever (the
@@ -2166,6 +2256,18 @@
 				</Popover>
 			{/if}
 			{#if mode === 'edit' && drafts.size > 0}
+				<!-- Draft autosave status for the whole pipeline bundle. Distinct
+				     from "Save all", which DEPLOYS the drafts — this only reflects
+				     that in-flight edits are persisted to the per-user server draft. -->
+				{#if $workspaceStore}
+					<AutosaveIndicator
+						workspace={$workspaceStore}
+						itemKind="data_pipeline"
+						path={pipelineDraftPath}
+						draftOnly
+						loadedFromDraft={loadedFromDbDraft}
+					/>
+				{/if}
 				<Button
 					variant="accent"
 					unifiedSize="sm"
@@ -2219,9 +2321,11 @@
 						<AssetGraphCanvas
 							graph={displayGraph}
 							selection={effectiveSelection}
+							hoveredPaths={activityHoverPaths}
+							selectedRunPaths={activitySelectPaths}
 							{activeRunnable}
 							activeRunnableIds={activeRunnables.ids}
-							runStates={activeRunnables.states}
+							runStates={mergedRunStates}
 							{pathPrefix}
 							defaultPathSuffix={DEFAULT_PATH_SUFFIX}
 							onCreateMissingTrigger={mode === 'edit' ? openMissingTriggerDrawer : undefined}
@@ -2279,11 +2383,14 @@
 							     here. -->
 							<PipelineActivityPanel
 								events={activityEvents}
+								edges={pipelineHistory.edges}
 								loading={pipelineHistory.loading}
 								truncated={pipelineHistory.truncated}
 								error={pipelineHistory.error}
 								days={activityDays}
 								onDaysChange={setActivityDays}
+								onHoverRun={(p) => (activityHoverPaths = p ?? [])}
+								onSelectRun={(p) => (activitySelectPaths = p ?? [])}
 							/>
 						{:else}
 							<AssetGraphDetailsPane
@@ -2333,6 +2440,7 @@
 								workspace={$workspaceStore}
 								onAnnotationsChange={handleAnnotationsChange}
 								onAssetsChange={handleAssetsChange}
+								onContentChange={handleContentChange}
 								onDraftPersist={handleDraftPersist}
 								onclose={() => {
 									// Close dismisses the pane but preserves drafts so