fix(frontend): group live pipeline runs in the activity panel (#9684)

* fix(frontend): refetch pipeline dispatch edges on live runs so cascades group

The activity panel groups runs into cascades by connected components of the
dispatch-edge graph, but edges came only from the one-shot history preload
(refetched on workspace/folder/days change). dispatch_event rows are written
server-side when a producer completes, so a run launched live had its producer
and freshly-dispatched children appear as live poll events with no connecting
edge — they rendered as separate ungrouped rows instead of one cascade.

Add an edges-only refetch and trigger it whenever the live poll's event id-set
changes, so live cascades converge to grouped like historic/scheduled ones.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(frontend): address review nits on live dispatch-edge refetch

- Sequence same-scope edges-only refetches with a monotonic edgeSeq so a
  slower earlier response can't overwrite a newer one mid-cascade (the gen
  counter only guards scope changes).
- Condense the duplicated edge-refetch rationale: keep the canonical "why"
  in loadEdges, trim the page effect comment to its trigger/loop invariant.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts

@@ -43,6 +43,10 @@ export function usePipelineHistory(
 	// Generation counter: a folder/days change mid-flight must not write a
 	// stale response into the new scope (mirrors the page's bodyFetchGen).
 	let gen = 0
+	// Edges-only refetches fire one-per-new-id during a cascade; `gen` only
+	// guards scope changes, so this sequences same-scope calls — a slower
+	// earlier response must not overwrite a newer one.
+	let edgeSeq = 0
 
 	async function load(ws: string, prefix: string, days: number) {
 		const myGen = ++gen
@@ -105,6 +109,31 @@ export function usePipelineHistory(
 		}
 	}
 
+	// Edges-only refetch (one cheap query, no completed-jobs paging): the
+	// `dispatch_event` rows that group a cascade are written server-side when a
+	// producer completes, so a run launched live has none in the one-shot
+	// preload. The page calls this when the live poll surfaces new jobs.
+	async function loadEdges(ws: string, prefix: string, days: number) {
+		// Skip while a full load owns `edges`; that load sets fresher edges and
+		// a concurrent write here could clobber it with a slightly older window.
+		if (loading) return
+		const myGen = gen
+		const mySeq = ++edgeSeq
+		const cutoff = new Date(Date.now() - days * 24 * 3600 * 1000).toISOString()
+		try {
+			const edgeRows = (await JobService.listAssetDispatchEdges({
+				workspace: ws,
+				pathStart: prefix,
+				createdAfter: cutoff
+			})) as DispatchEdge[]
+			// Drop a stale-scope write, or one a newer refetch already superseded.
+			if (gen !== myGen || mySeq !== edgeSeq) return
+			edges = edgeRows
+		} catch (e) {
+			console.warn('failed to refetch pipeline dispatch edges', e)
+		}
+	}
+
 	$effect(() => {
 		const ws = getWorkspace()
 		const prefix = getPathPrefix()
@@ -142,6 +171,13 @@ export function usePipelineHistory(
 			const prefix = getPathPrefix()
 			if (!ws || !prefix || !getEnabled()) return
 			void load(ws, prefix, getDays())
+		},
+		/** Re-pull just the dispatch edges (see `loadEdges`). */
+		refetchEdges() {
+			const ws = getWorkspace()
+			const prefix = getPathPrefix()
+			if (!ws || !prefix || !getEnabled()) return
+			void loadEdges(ws, prefix, getDays())
 		}
 	}
 }

frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte

@@ -1589,6 +1589,21 @@
 		for (const e of activeRunnables.events) byId.set(e.id, e)
 		return Array.from(byId.values()).sort((a, b) => b.at.localeCompare(a.at))
 	})
+	// Keep dispatch edges live so freshly-launched runs group (see `loadEdges`).
+	// The poll's id set changes the instant a new job appears (a dispatched
+	// child is a new id) — exactly when fresh edges exist — so re-pull then.
+	// Keyed on ids, not status, so queued→done ticks don't refetch;
+	// `lastLiveEventSig` is a plain `let` so writing it can't retrigger this.
+	let lastLiveEventSig = ''
+	$effect(() => {
+		const sig = activeRunnables.events
+			.map((e) => e.id)
+			.sort()
+			.join(',')
+		if (sig === lastLiveEventSig) return
+		lastLiveEventSig = sig
+		pipelineHistory.refetchEdges()
+	})
 	// 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