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>
This commit is contained in:
Ruben Fiszel
2026-05-15 16:23:49 +00:00
co-authored by Claude Opus 4.7
parent b37e6830fe
commit 9f0c0258b3
5 changed files with 314 additions and 15 deletions
@@ -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>
@@ -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 })
})
@@ -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,
@@ -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()
}
}
}
@@ -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`