From 1be4df9acb935250d4cc12e83cf67e366d870d5a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 19 Jun 2026 16:43:23 +0200 Subject: [PATCH] fix(frontend): group live pipeline runs in the activity panel (#9684) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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) * 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) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../AssetGraph/pipelineHistory.svelte.ts | 36 +++++++++++++++++++ .../(logged)/pipeline/[folder]/+page.svelte | 15 ++++++++ 2 files changed, 51 insertions(+) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts index 6d69c564ab..50ad3e1d6e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts +++ b/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()) } } } diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index ff3208493c..c98b11139f 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/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