From 9f0c0258b3c9accb71e1f684bb8f665deba7fdf9 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Fri, 15 May 2026 16:23:49 +0000
Subject: [PATCH] feat: live run activity + status badges in pipeline asset
 graph

- folder-scoped queue poll lights up the downstream asset-trigger
  cascade (not just the launched script); zero requests at rest,
  catch-up for fast hops, auto-disarm when idle
- per-runnable node badge: last-run status + session run count
- animate unsaved/live-parsed edges (was unconditionally suppressed)
- background-pane click no longer clears selection
- run-bridge guarded so node selection/save no longer triggers a test

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../assets/AssetGraph/AssetGraphCanvas.svelte |  48 +++--
 .../AssetGraph/AssetGraphDetailsPane.svelte   |  39 +++-
 .../assets/AssetGraph/RunnableNode.svelte     |  37 ++++
 .../AssetGraph/activeRunnables.svelte.ts      | 184 ++++++++++++++++++
 .../(logged)/pipeline/[folder]/+page.svelte   |  21 ++
 5 files changed, 314 insertions(+), 15 deletions(-)
 create mode 100644 frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts

diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
index 9cf36f4123..b8e808dadb 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
@@ -16,6 +16,7 @@
 	import AssetGraphEdge from './AssetGraphEdge.svelte'
 	import { layoutAssetGraph } from './assetGraphLayout'
 	import type { AssetGraphResponse, AssetGraphSelection } from './types'
+	import type { RunnableRunState } from './activeRunnables.svelte'
 	import type { AssetKind } from '$lib/gen'
 	import { NODE } from '$lib/components/graph/util'
 
@@ -91,9 +92,19 @@
 		}) => void
 		// Runnable currently executing — its incoming asset/trigger edges and
 		// outgoing write edges get an animated stroke to convey "this is
-		// running right now". Cleared by the page when the run reaches a
-		// terminal state. Future: same hook for live pipeline status.
+		// running right now". Zero-latency hint for the script the user just
+		// launched (set/cleared by the page from the in-pane Test state).
 		activeRunnable?: { kind: 'script' | 'flow'; path: string } | undefined
+		// `${kind}:${path}` ids of every pipeline runnable with an in-flight
+		// (or just-finished) job, from the folder-scoped queue poll. This is
+		// what lights up the *downstream cascade* (jobs the user didn't launch
+		// directly). Merged with `activeRunnable` so the launched script lights
+		// instantly while cascade hops light as the poll observes them.
+		activeRunnableIds?: ReadonlySet<string>
+		// `${kind}:${path}` → last-run status + session run count, rendered as
+		// a small badge on each runnable node. Same poll source as
+		// `activeRunnableIds`; persists the last status while idle.
+		runStates?: ReadonlyMap<string, RunnableRunState>
 	}
 	let {
 		graph,
@@ -106,7 +117,9 @@
 		defaultScheduleCron = '',
 		onRunProducer,
 		onRunnableMenuRemove,
-		activeRunnable
+		activeRunnable,
+		activeRunnableIds,
+		runStates
 	}: Props = $props()
 
 	const ADD_NODE_ID = '__add__'
@@ -248,6 +261,7 @@
 									})
 							: undefined,
 					downstreamCount: downstreamByScript.get(r.path) ?? 0,
+					runState: runStates?.get(`${r.usage_kind}:${r.path}`),
 					onSelectSelf: () =>
 						onselect?.({
 							kind: 'runnable',
@@ -415,11 +429,16 @@
 		})
 	})
 
-	// Runnable graph-id, if a job is currently running for it. Used below
-	// to animate the edges into and out of that node — keeps the static
-	// graph quiet and reserves animation as an "is happening now" signal.
-	let activeRunnableId = $derived(
-		activeRunnable ? `${activeRunnable.kind}:${activeRunnable.path}` : undefined
+	// Graph-ids of every runnable that's executing right now: the polled
+	// in-flight/just-finished set (cascade) plus the zero-latency hint for
+	// the script the user just launched. Edges into/out of any of these get
+	// the animated stroke — animation stays reserved as an "is happening
+	// now" signal and the static graph stays quiet when nothing runs.
+	let activeRunnableIdSet = $derived(
+		new Set<string>([
+			...(activeRunnableIds ?? []),
+			...(activeRunnable ? [`${activeRunnable.kind}:${activeRunnable.path}`] : [])
+		])
 	)
 	let flowEdges = $derived.by(() =>
 		model.edges
@@ -427,8 +446,8 @@
 			.filter((e) => e.kind !== 'add-anchor')
 			.map<Edge>((e) => {
 				const touchesActiveRunnable =
-					activeRunnableId !== undefined &&
-					(e.source === activeRunnableId || e.target === activeRunnableId)
+					activeRunnableIdSet.size > 0 &&
+					(activeRunnableIdSet.has(e.source) || activeRunnableIdSet.has(e.target))
 				let style: string
 				let animated = false
 				let markerColor: string | undefined = undefined
@@ -476,10 +495,16 @@
 				// parsed, not persisted. Lineage edges from base graph data
 				// don't carry `unsaved`, only those synthesized by the draft
 				// overlay (e.g. the random output asset).
+				//
+				// `animated` is intentionally NOT cleared here: a live-parsed
+				// output edge still belongs to a script that can be run, and
+				// the run feedback (flow animation toward its output assets)
+				// is exactly what the user expects while it executes. The
+				// dashed+dimmed treatment already conveys "not persisted";
+				// it composes fine with the active-run animation.
 				if (e.unsaved) {
 					strokeDasharray = '3 3'
 					style = `${style} opacity: 0.7;`
-					animated = false
 					if (label) label = `${label} (unsaved)`
 				}
 				if (strokeDasharray) {
@@ -553,7 +578,6 @@
 		defaultEdgeOptions={{ type: 'asset' }}
 		proOptions={{ hideAttribution: true }}
 		onnodeclick={handleNodeClick}
-		onpaneclick={() => onselect?.(undefined)}
 		--background-color={false}
 	>
 		<div class="absolute inset-0 !bg-surface-secondary h-full"></div>
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
index b84e34215c..0e63d05fb0 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
@@ -190,17 +190,50 @@
 	// runs initiated from the graph, not just the in-pane Test button.
 	let scriptEditorRef: { runTest: (opts?: { cascade?: boolean }) => Promise<unknown> } | undefined =
 		$state(undefined)
+	// Run-bridge: fire `ScriptEditor.runTest()` exactly once per distinct
+	// signal-counter value the page sends (canvas Run targeting the open
+	// script), on the editor that actually corresponds to the run target.
+	//   1. A plain node selection or a save must never trigger a run — they
+	//      bump no counter, so the last-handled guard keeps the effect inert
+	//      even though selecting/saving reassigns `scriptEditorRef`.
+	//   2. The Run button selects the node first, so the ScriptEditor is
+	//      usually still (re)mounting when the counter bumps. We must NOT
+	//      run yet — and must NOT mark the signal handled — so the effect
+	//      re-runs and fires once the editor settles.
+	//   3. `scriptRes` is stale-while-revalidate: right after selection the
+	//      PREVIOUS script's ScriptEditor is still mounted. Firing then runs
+	//      the wrong script. `editorReadyForTarget` gates on the mounted
+	//      editor being settled on the run-target path before we fire.
+	let lastHandledRunSig = 0
+	let lastHandledCascadeSig = 0
+	let runTargetPath = $derived.by(() =>
+		isDraft
+			? draftScript?.path
+			: selection?.kind === 'runnable' && selection.runnable_kind === 'script'
+				? selection.path
+				: undefined
+	)
+	let editorReadyForTarget = $derived.by(
+		() =>
+			!!scriptEditorRef &&
+			runTargetPath !== undefined &&
+			script?.path === runTargetPath &&
+			(isDraft || (!scriptRes.loading && scriptRes.current?.path === runTargetPath))
+	)
 	$effect(() => {
-		// Track the counter; ignore the initial 0/undefined.
 		const sig = requestRunSignal
-		if (sig === undefined || sig === 0) return
+		if (sig === undefined || sig === 0 || sig === lastHandledRunSig) return
+		if (!editorReadyForTarget) return
+		lastHandledRunSig = sig
 		void scriptEditorRef?.runTest()
 	})
 	$effect(() => {
 		// Cascade-explicit sister signal: same pattern, but forces the
 		// runTest call to opt into the asset-trigger cascade for this run.
 		const sig = requestRunCascadeSignal
-		if (sig === undefined || sig === 0) return
+		if (sig === undefined || sig === 0 || sig === lastHandledCascadeSig) return
+		if (!editorReadyForTarget) return
+		lastHandledCascadeSig = sig
 		void scriptEditorRef?.runTest({ cascade: true })
 	})
 
diff --git a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte
index b562464259..bd283edd64 100644
--- a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte
@@ -1,6 +1,7 @@
 <script lang="ts">
 	import { Handle, Position } from '@xyflow/svelte'
 	import {
+		CheckCircle2,
 		ChevronDown,
 		Code2,
 		EllipsisVertical,
@@ -10,11 +11,13 @@
 		Play,
 		Timer,
 		Trash2,
+		XCircle,
 		Zap
 	} from 'lucide-svelte'
 	import { twMerge } from 'tailwind-merge'
 	import { preventDefault, stopPropagation } from 'svelte/legacy'
 	import type { GraphUsageKind } from './types'
+	import type { RunnableRunState } from './activeRunnables.svelte'
 	import { NODE } from '$lib/components/graph/util'
 	import DropdownV2 from '$lib/components/DropdownV2.svelte'
 	import Popover from '$lib/components/meltComponents/Popover.svelte'
@@ -29,6 +32,9 @@
 			in_pipeline?: boolean
 			partition_kind?: 'daily' | 'hourly' | 'weekly' | 'monthly' | 'dynamic'
 			freshness?: string
+			// Last-run status + run count observed this session (from the
+			// folder queue poll). Undefined until the first observed run.
+			runState?: RunnableRunState
 			// True for nodes synthesized from local drafts (script not yet
 			// persisted). Same convention as `unsaved` on triggers/edges.
 			unsaved?: boolean
@@ -167,6 +173,37 @@
 				<span class="text-3xs leading-none">{data.freshness}</span>
 			</div>
 		{/if}
+		{#if data.runState}
+			{@const rs = data.runState}
+			<div
+				class={twMerge(
+					'shrink-0 flex items-center gap-0.5 px-1 py-0.5 mr-1 rounded-sm',
+					rs.status === 'running'
+						? 'bg-blue-100 dark:bg-blue-900/40 text-blue-700 dark:text-blue-300'
+						: rs.status === 'success'
+							? 'bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300'
+							: 'bg-red-100 dark:bg-red-900/40 text-red-700 dark:text-red-300'
+				)}
+				title={`${
+					rs.status === 'running'
+						? 'Running now'
+						: rs.status === 'success'
+							? 'Last run succeeded'
+							: 'Last run failed'
+				}${rs.runs > 0 ? ` — ran ${rs.runs}× this session` : ''}`}
+			>
+				{#if rs.status === 'running'}
+					<Loader2 size={10} class="animate-spin" />
+				{:else if rs.status === 'success'}
+					<CheckCircle2 size={10} />
+				{:else}
+					<XCircle size={10} />
+				{/if}
+				{#if rs.runs > 0}
+					<span class="text-3xs leading-none tabular-nums">×{rs.runs}</span>
+				{/if}
+			</div>
+		{/if}
 	</div>
 	{#if showRun}
 		<!-- Run button revealed on hover/select. Matches the placement, size,
diff --git a/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts
new file mode 100644
index 0000000000..4698ebf447
--- /dev/null
+++ b/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts
@@ -0,0 +1,184 @@
+import { JobService } from '$lib/gen'
+
+export type RunStatus = 'running' | 'success' | 'failure'
+/** Per-runnable badge state: latest run status + runs observed this session. */
+export type RunnableRunState = { status: RunStatus; runs: number }
+
+/**
+ * Tracks which pipeline runnables currently have an in-flight (queued/running)
+ * job — plus any that started AND finished since the last poll, so a fast
+ * cascade hop that completes between two ticks still pulses once. Returns a
+ * reactive `Set<string>` of `${kind}:${path}` ids (matching the graph's
+ * runnable node ids) for the canvas to animate.
+ *
+ * Cost model — deliberately cheap:
+ *  - ZERO requests while idle. Polling only runs after `arm()` (a user-
+ *    initiated run) and self-sustains only while jobs are in flight.
+ *  - One folder-scoped, `perPage`-capped `listExtendedJobs` request per tick.
+ *  - Auto-disarms after `MAX_IDLE_TICKS` consecutive empty ticks (rides the
+ *    brief gaps between cascade hops), with a hard total-duration cap.
+ *
+ * Read-only (`listExtendedJobs` is a read endpoint, no editor dependency), so
+ * a future read-only / operator pipeline view can reuse this as-is.
+ */
+export function useActiveRunnableIds(
+	getWorkspace: () => string | undefined,
+	getPathPrefix: () => string | undefined
+) {
+	const INTERVAL_MS = 3000
+	const MAX_IDLE_TICKS = 3 // ~9s of no in-flight jobs before disarming
+	const HARD_CAP_MS = 5 * 60_000 // safety: stop a runaway poll loop
+	const PER_PAGE = 50 // folder-scoped + newest-first: plenty for a cascade
+	// Small look-back so a job that finished right after the previous tick
+	// fired (but before its response landed) isn't dropped from catch-up.
+	const CATCHUP_OVERLAP_MS = 4000
+
+	let ids = $state(new Set<string>())
+	// Per-runnable last-run state + session run count, for the node badge.
+	// Reassigned as a fresh Map each tick (reactive without SvelteMap).
+	let states = $state(new Map<string, RunnableRunState>())
+	let timer: ReturnType<typeof setTimeout> | undefined
+	let running = false
+	let idleTicks = 0
+	let startedAtMs = 0
+	let lastPollTs = new Date(0).toISOString()
+
+	// Persistent (non-reactive) accumulators — survive `stop()` so the badge
+	// keeps showing the last status while idle; only `dispose()` clears them.
+	const completedHistory = new Map<
+		string,
+		{ runs: number; lastStatus: RunStatus; lastTs: string }
+	>()
+	const countedJobIds = new Set<string>()
+
+	function idOf(jobKind: string | undefined, path: string | undefined): string | undefined {
+		if (!path || !jobKind) return undefined
+		const kind = jobKind.startsWith('flow') ? 'flow' : 'script'
+		return `${kind}:${path}`
+	}
+
+	function stop() {
+		if (timer !== undefined) clearTimeout(timer)
+		timer = undefined
+		running = false
+		idleTicks = 0
+		if (ids.size > 0) ids = new Set()
+	}
+
+	async function tick() {
+		const ws = getWorkspace()
+		const prefix = getPathPrefix()
+		if (!ws || !prefix) {
+			stop()
+			return
+		}
+		const since = lastPollTs
+		const pollStartedMs = Date.now()
+		let next = new Set<string>()
+		const inFlightThisTick = new Set<string>()
+		let anyInFlight = false
+		try {
+			const res = await JobService.listExtendedJobs({
+				workspace: ws,
+				scriptPathStart: prefix,
+				jobKinds: ['preview', 'script', 'flow', 'flowpreview', 'flowscript'].join(','),
+				perPage: PER_PAGE,
+				page: 1
+			})
+			for (const j of res.jobs ?? []) {
+				const id = idOf((j as any).job_kind, (j as any).script_path)
+				if (!id) continue
+				if ((j as any).type === 'QueuedJob') {
+					// queued or running — currently active
+					next.add(id)
+					inFlightThisTick.add(id)
+					anyInFlight = true
+				} else {
+					// completed: catch-up — a hop that started after the last
+					// poll (and already finished) still gets one pulse.
+					const started: string | undefined = (j as any).started_at ?? (j as any).created_at
+					if (started && started >= since) next.add(id)
+					// Tally distinct completed jobs for the node badge. Jobs
+					// come newest-first; only let a strictly newer completion
+					// move the displayed status.
+					const jobId: string | undefined = (j as any).id
+					if (jobId && !countedJobIds.has(jobId)) {
+						countedJobIds.add(jobId)
+						const prev = completedHistory.get(id)
+						const status: RunStatus = (j as any).success === true ? 'success' : 'failure'
+						const ts = started ?? new Date(pollStartedMs).toISOString()
+						completedHistory.set(id, {
+							runs: (prev?.runs ?? 0) + 1,
+							lastStatus: !prev || ts >= prev.lastTs ? status : prev.lastStatus,
+							lastTs: !prev || ts >= prev.lastTs ? ts : prev.lastTs
+						})
+					}
+				}
+			}
+		} catch {
+			// Transient failure: keep the previous set, retry next tick.
+			next = ids
+			anyInFlight = ids.size > 0
+		}
+		ids = next
+		// Rebuild the badge snapshot: in-flight wins (spinner), otherwise the
+		// last completed status. Fresh Map each tick → reactive without
+		// SvelteMap. completedHistory persists across `stop()`.
+		const snap = new Map<string, RunnableRunState>()
+		for (const [id, h] of completedHistory) {
+			snap.set(id, { status: inFlightThisTick.has(id) ? 'running' : h.lastStatus, runs: h.runs })
+		}
+		for (const id of inFlightThisTick) {
+			if (!snap.has(id)) snap.set(id, { status: 'running', runs: 0 })
+		}
+		states = snap
+		// Look back slightly so a job finishing in the request window isn't
+		// missed by the next tick's catch-up comparison.
+		lastPollTs = new Date(pollStartedMs - CATCHUP_OVERLAP_MS).toISOString()
+
+		if (anyInFlight) idleTicks = 0
+		else idleTicks++
+
+		const expired = Date.now() - startedAtMs > HARD_CAP_MS
+		if (idleTicks > MAX_IDLE_TICKS || expired) {
+			stop()
+			return
+		}
+		timer = setTimeout(() => void tick(), INTERVAL_MS)
+	}
+
+	return {
+		get ids() {
+			return ids
+		},
+		/** Per-runnable last-run status + session run count, for node badges. */
+		get states() {
+			return states
+		},
+		/**
+		 * Kick the poll loop (call when a run is launched from this view).
+		 * Idempotent: if already polling, just resets the idle counter so the
+		 * loop keeps going to cover the freshly-launched run + its cascade.
+		 */
+		arm() {
+			idleTicks = 0
+			if (running) return
+			running = true
+			startedAtMs = Date.now()
+			// Open the catch-up window from now so we don't replay old history.
+			lastPollTs = new Date(Date.now() - CATCHUP_OVERLAP_MS).toISOString()
+			void tick()
+		},
+		/**
+		 * Stop polling and fully reset (call on destroy / folder change).
+		 * Unlike `stop()` this also wipes the persisted badge history so a
+		 * different folder doesn't inherit the previous one's run state.
+		 */
+		dispose() {
+			stop()
+			completedHistory.clear()
+			countedJobIds.clear()
+			if (states.size > 0) states = new Map()
+		}
+	}
+}
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
index 7c64644f8f..052a585168 100644
--- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
@@ -5,6 +5,7 @@
 	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 AssetGraphDetailsPane from '$lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte'
 	import PipelinePickerModal from '$lib/components/assets/AssetGraph/PipelinePickerModal.svelte'
 	import type { AssetWithAltAccessType } from '$lib/components/assets/lib'
@@ -917,6 +918,20 @@
 	// AssetRunsPanel reports the run is done. The same hook will later
 	// be reused for live pipeline status.
 	let activeRunnable = $state<{ kind: 'script' | 'flow'; path: string } | undefined>(undefined)
+	// Folder-scoped poll of in-flight (and just-finished) pipeline jobs. This
+	// is what lights up the *downstream cascade* on the canvas — jobs the user
+	// didn't launch directly. `arm()` is called when a run is launched from
+	// this view; it polls only while jobs are in flight and stops when idle
+	// (zero requests at rest). Re-scoped/torn down when the folder changes or
+	// the page unmounts.
+	const activeRunnables = useActiveRunnableIds(
+		() => $workspaceStore,
+		() => pathPrefix
+	)
+	$effect(() => {
+		pathPrefix // re-scope poll to the current folder
+		return () => activeRunnables.dispose()
+	})
 	// 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
@@ -1215,6 +1230,8 @@
 								graph={graphWithDraft}
 								selection={effectiveSelection}
 								{activeRunnable}
+								activeRunnableIds={activeRunnables.ids}
+								runStates={activeRunnables.states}
 								{pathPrefix}
 								defaultPathSuffix={DEFAULT_PATH_SUFFIX}
 								defaultScheduleCron={DEFAULT_SCHEDULE_CRON}
@@ -1286,6 +1303,10 @@
 									// immediately — its background poll only kicks in
 									// for already-listed in-flight jobs.
 									if (!$workspaceStore || producer.kind !== 'script') return undefined
+									// Start the folder-scoped poll so the downstream
+									// asset-trigger cascade (jobs not launched here)
+									// lights up its edges as it fans out.
+									activeRunnables.arm()
 									// Cascade default: same as the Test button — `cascade`
 									// undefined / false skips the asset-trigger dispatch
 									// via `_wmill_skip_asset_dispatch`; explicit `true`