Broadcast workload frames in order and keep same-numbered jobs apart

desktop/scripts/verify-service-contracts.ts

@@ -137,8 +137,11 @@ const NOTIFY_LITERAL_RE =
     /(?:(?:notify|emit|publish|broadcast)\(\s*|notifyWithShutdown\(\s*ctx\s*,\s*)"([^"]+)"/gi
 const NOTIFY_PREFIX_CONCAT_RE =
     /(?:(?:notify|emit|publish|broadcast)\(\s*|notifyWithShutdown\(\s*ctx\s*,\s*)"([^"]+)"\s*\+/i
+// Broadcast(ctx, frame) is HTTP fan-out, not a JSON-RPC notification. Keep
+// literal Broadcast methods above, but do not treat an identifier argument to
+// Broadcast as an unresolved JSON-RPC method.
 const NOTIFY_IDENT_RE =
-    /(?:(?:notify|emit|publish|broadcast)\(\s*|notifyWithShutdown\(\s*ctx\s*,\s*)([A-Za-z_]\w*)\s*,/gi
+    /(?:(?:notify|emit|publish)\(\s*|notifyWithShutdown\(\s*ctx\s*,\s*)([A-Za-z_]\w*)\s*,/gi
 
 /** A `| `nvpair-ui-broker` | 0.37.0 | … |` row — forbidden in hand-maintained docs. */
 const DOC_COMPONENT_ROW_RE = /^\|\s*`([a-z0-9-]+)`\s*\|\s*(\d+\.\d+\.\d+)\s*\|/gm

desktop/src/electron/service-bridge/modular-state.ts

@@ -38,6 +38,7 @@ import type {
     Workload,
     WorkloadLiveEngine,
     WorkloadLiveProgress,
+    WorkloadRemoval,
     WorkloadState
 } from '@/shared/types/workloads'
 import type { WorkerState, WorkerStatus } from '@/shared/types/workers'
@@ -55,7 +56,7 @@ import {
     parseEngineCapabilities
 } from '@/shared/utils/engines'
 import { engineProgressKey } from '@/shared/utils/engine-progress'
-import { workloadKey } from '@/shared/utils/workloads'
+import { workloadKey, workloadMatchesRemoval } from '@/shared/utils/workloads'
 import { currentPlatform, platformDisplayName } from '@/shared/utils/platform'
 import { emitBridgePush } from './broadcaster'
 import { createStructuredLogger } from '@/shared/utils/log'
@@ -673,6 +674,7 @@ function parseWorkload(value: JsonValue | undefined): Workload | null {
         engine,
         state: stateValue,
         originatedFrom: nullableStringValue(obj.originatedFrom),
+        runId: nullableStringValue(obj.runId),
         createdAt: numberValue(obj.createdAt),
         startedAt: nullableNumberValue(obj.startedAt),
         completedAt: nullableNumberValue(obj.completedAt),
@@ -1520,6 +1522,13 @@ class ModularBridgeState {
     private workloads = new Map<string, Workload>()
     // Exists only during one epoch snapshot fetch, including removal tombstones.
     private workloadRecovery: Set<string> | null = null
+    /**
+     * Broker removals seen while a recovery snapshot is fetched. A removal
+     * names a job by `(originatedFrom, id)` only, so it is kept as itself
+     * rather than as keys: it must also retire a matching job that is not in
+     * the catalog yet but is in the older snapshot.
+     */
+    private workloadRecoveryRemovals: WorkloadRemoval[] = []
     private logs: LogEntry[] = []
     // Per-engine bound proxy port reported by the broker. 0 = not reported yet;
     // we never fabricate a default — an unknown port surfaces as null, not a
@@ -1948,7 +1957,7 @@ class ModularBridgeState {
      */
     seedWorkloads(workloads: Workload[]): WsInvokeResponse<'workloads:get-initial'> {
         for (const workload of workloads) {
-            const key = workloadKey(workload.originatedFrom, workload.id)
+            const key = workloadKey(workload)
             if (!this.workloads.has(key)) this.workloads.set(key, workload)
         }
         return this.getWorkloads()
@@ -1958,25 +1967,27 @@ class ModularBridgeState {
     beginWorkloadRecovery(): Set<string> {
         const snapshot = new Set<string>()
         this.workloadRecovery = snapshot
+        this.workloadRecoveryRemovals = []
         return snapshot
     }
 
     /** Apply an authoritative baseline, or just release an invalidated/failed fetch. */
     endWorkloadRecovery(snapshot: Set<string>, workloads?: Workload[]): void {
         if (this.workloadRecovery !== snapshot) return
         this.workloadRecovery = null
+        const removals = this.workloadRecoveryRemovals
+        this.workloadRecoveryRemovals = []
         if (!workloads) {
             snapshot.clear()
             return
         }
-        const baseline = new Map(
-            workloads.map(workload => [workloadKey(workload.originatedFrom, workload.id), workload])
-        )
+        const baseline = new Map(workloads.map(workload => [workloadKey(workload), workload]))
         for (const [key, workload] of this.workloads) {
             if (!snapshot.has(key) && !baseline.has(key)) this.evictWorkload(key, workload)
         }
         for (const [key, workload] of baseline) {
             if (snapshot.has(key)) continue
+            if (removals.some(removal => workloadMatchesRemoval(workload, removal))) continue
             this.workloads.set(key, workload)
             emitBridgePush('workloads:upsert', workload)
         }
@@ -1988,22 +1999,29 @@ class ModularBridgeState {
         const obj = objectValue(params)
         const workload = parseWorkload(obj?.workloadInfo)
         if (!workload) return
-        const key = workloadKey(workload.originatedFrom, workload.id)
+        const key = workloadKey(workload)
         this.workloadRecovery?.add(key)
         this.workloads.set(key, workload)
         emitBridgePush('workloads:upsert', workload)
     }
 
-    /** Relay a broker `workloads:remove` (`{ workloadId, originatedFrom? }`) into the catalog + UI. */
+    /**
+     * Relay a broker `workloads:remove` (`{ workloadId, originatedFrom? }`) into
+     * the catalog + UI. It names no engine and no run, so, like the broker's own
+     * store, it drops every job matching the pair.
+     */
     removeWorkloadFromParams(params: JsonValue | undefined): void {
         const obj = objectValue(params)
         const workloadId = stringValue(obj?.workloadId)
         if (!workloadId) return
-        const originatedFrom = nullableStringValue(obj?.originatedFrom)
-        const key = workloadKey(originatedFrom, workloadId)
-        this.workloadRecovery?.add(key)
-        this.workloads.delete(key)
-        emitBridgePush('workloads:remove', { workloadId, originatedFrom })
+        const removal = { workloadId, originatedFrom: nullableStringValue(obj?.originatedFrom) }
+        if (this.workloadRecovery) this.workloadRecoveryRemovals.push(removal)
+        for (const [key, workload] of this.workloads) {
+            if (!workloadMatchesRemoval(workload, removal)) continue
+            this.workloadRecovery?.add(key)
+            this.workloads.delete(key)
+        }
+        emitBridgePush('workloads:remove', removal)
     }
 
     /**
@@ -2027,7 +2045,9 @@ class ModularBridgeState {
         this.workloads.delete(key)
         emitBridgePush('workloads:remove', {
             workloadId: workload.id,
-            originatedFrom: workload.originatedFrom
+            originatedFrom: workload.originatedFrom,
+            engine: workload.engine,
+            runId: workload.runId ?? null
         })
     }
 

desktop/src/electron/tray-badge.ts

@@ -26,7 +26,7 @@ import {
     type TrayBadgeInk,
     type TrayBadgeMap
 } from '@/shared/utils/tray-badge-map'
-import { workloadKey } from '@/shared/utils/workloads'
+import { workloadKey, workloadMatchesRemoval } from '@/shared/utils/workloads'
 
 export interface TrayBadgeSink {
     setIcon: (map: TrayBadgeMap) => void
@@ -66,7 +66,7 @@ export function startTrayBadge(
 ): () => void {
     const catalog = new Map<string, Workload>()
     for (const workload of initial) {
-        catalog.set(workloadKey(workload.originatedFrom, workload.id), workload)
+        catalog.set(workloadKey(workload), workload)
     }
 
     let signature = ''
@@ -105,9 +105,11 @@ export function startTrayBadge(
 
     const unsubscribe = subscribePush(event => {
         if (event.channel === 'workloads:upsert') {
-            catalog.set(workloadKey(event.payload.originatedFrom, event.payload.id), event.payload)
+            catalog.set(workloadKey(event.payload), event.payload)
         } else if (event.channel === 'workloads:remove') {
-            catalog.delete(workloadKey(event.payload.originatedFrom, event.payload.workloadId))
+            for (const [key, workload] of catalog) {
+                if (workloadMatchesRemoval(workload, event.payload)) catalog.delete(key)
+            }
         } else {
             return
         }

desktop/src/shared/types/workloads.ts

@@ -13,11 +13,17 @@ export interface Workload {
     state: WorkloadState
     /**
      * Owner/origin node of the workload — the node whose proxy received the
-     * request. This is the identity half of the backend's `(originatedFrom, id)`
-     * global catalog key (workload ids are a per-node proxy counter, so they
-     * collide across nodes; `originatedFrom` disambiguates).
+     * request. Part of the backend's `(originatedFrom, engine, runId, id)`
+     * catalog key: workload ids are a per-proxy counter, so they collide across
+     * nodes and engines.
      */
     originatedFrom: string | null
+    /**
+     * The proxy run that numbered this job. A restarted proxy counts from 1
+     * again; the run id tells its jobs from the earlier run's. Absent from a
+     * producer that predates it.
+     */
+    runId?: string | null
     /**
      * Node the workload was routed to / scheduled on — where it actually ran,
      * as opposed to `originatedFrom` (where it came from). Optional/additive:
@@ -131,3 +137,11 @@ export interface WorkloadLiveProgress {
      */
     engineProgress?: WorkloadLiveEngine | null
 }
+
+/** A `workloads:remove` push. The broker's carries no engine and no run id. */
+export interface WorkloadRemoval {
+    workloadId: string
+    originatedFrom: string | null
+    engine?: EngineType | null
+    runId?: string | null
+}

desktop/src/shared/types/ws-channels.ts

@@ -39,7 +39,7 @@ import type {
     NodeItemMetrics,
     NodeInferenceMetricsSnapshot
 } from '@/shared/types/metrics'
-import type { Workload, WorkloadLiveProgress } from '@/shared/types/workloads'
+import type { Workload, WorkloadLiveProgress, WorkloadRemoval } from '@/shared/types/workloads'
 import type { WorkerStatus } from '@/shared/types/workers'
 import type {
     AvailableNode,
@@ -171,7 +171,7 @@ export interface WsPushChannelMap {
     // workload ids are a per-node proxy counter (the catalog is keyed by the
     // (originatedFrom, id) pair).
     'workloads:upsert': Workload
-    'workloads:remove': { workloadId: string; originatedFrom: string | null }
+    'workloads:remove': WorkloadRemoval
     /**
      * The broker's live view of a running job: what PAIR measured of the
      * client's stream (elapsed time, chunks, bytes, time to first byte) plus

desktop/src/shared/utils/workloads.ts

@@ -1,24 +1,53 @@
 // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-import type { Workload, WorkloadLiveProgress } from '@/shared/types/workloads'
+import type { Workload, WorkloadLiveProgress, WorkloadRemoval } from '@/shared/types/workloads'
 import type { EngineType } from '@/shared/types/engines'
 
+/** The fields that identify one job. */
+interface WorkloadIdentity {
+    originatedFrom: string | null
+    engine: EngineType | null
+    runId?: string | null
+    id: string
+}
+
 /**
- * Stable catalog key for a workload.
+ * Stable catalog key for a workload: the backend's own identity,
+ * `(originatedFrom, engine, runId, id)`.
  *
- * The backend's catalog is keyed by `(originatedFrom, engine, runId, id)`, but
- * the `workloads:remove` push carries only `(workloadId, originatedFrom)` — and
- * the broker's own `Store.Remove` drops every record matching that pair — so
- * `(originatedFrom, id)` is the only key a subscribe client can maintain
- * consistently across upsert and remove. Each node's proxy assigns workload ids
- * from its own monotonic counter, so ids collide across nodes; `originatedFrom`
- * (the origin node) disambiguates. Mirror that here so a remote node's job never
- * overwrites a local one that happens to share an id. The `\u0000` separator
- * cannot appear in a host id or proxy counter, so the key is unambiguous.
+ * Every proxy numbers its jobs from 1, and numbers again from 1 when it
+ * restarts. The origin tells nodes apart, the engine tells an Ollama job "3"
+ * from an oMLX job "3" on one node, and the run id tells a job from the one a
+ * restarted proxy gave the same number. Keyed on fewer fields, one job's card
+ * would overwrite another's. The `\u0000` separator cannot appear in any of
+ * the fields, so the key is unambiguous.
+ */
+export function workloadKey(workload: WorkloadIdentity): string {
+    return [
+        workload.originatedFrom ?? '',
+        workload.engine ?? '',
+        workload.runId ?? '',
+        workload.id
+    ].join('\u0000')
+}
+
+/**
+ * Whether a removal names this workload. A removal from the broker carries
+ * only `(workloadId, originatedFrom)` and, like the broker's own store, drops
+ * every job matching that pair. One raised by the desktop itself names the
+ * engine and run too — `runId: null` meaning "the job with no run", not "any
+ * run" — so it drops exactly one job. Only an absent field is a wildcard.
  */
-export function workloadKey(originatedFrom: string | null, id: string): string {
-    return `${originatedFrom ?? ''}\u0000${id}`
+export function workloadMatchesRemoval(
+    workload: WorkloadIdentity,
+    removal: WorkloadRemoval
+): boolean {
+    if (workload.id !== removal.workloadId) return false
+    if ((workload.originatedFrom ?? '') !== (removal.originatedFrom ?? '')) return false
+    if (removal.engine !== undefined && workload.engine !== removal.engine) return false
+    if (removal.runId !== undefined && (workload.runId ?? null) !== removal.runId) return false
+    return true
 }
 
 /**

desktop/src/ui/components/Workloads/WorkloadItemCard.tsx

@@ -4,7 +4,7 @@
 import { useMemo, memo } from 'react'
 import { Card, Flex, Stack, Text } from '@nvidia/foundations-react-core'
 import type { Workload } from '@/shared/types/workloads'
-import { workloadExecutionNodeId } from '@/shared/utils/workloads'
+import { workloadExecutionNodeId, workloadKey } from '@/shared/utils/workloads'
 import { useNodesStore } from '@/ui/stores/nodes.store'
 import { useWorkloadLiveProgress } from '@/ui/stores/workloads.store'
 import { formatModelDisplayName } from '@/ui/utils/format-model-display-name'
@@ -148,8 +148,7 @@ function WorkloadItemCard({ workload }: { workload: Workload }) {
             className={className}
             density="compact"
             style={{ direction: 'ltr' }}
-            data-workload-id={workload.id}
-            data-workload-origin={workload.originatedFrom ?? ''}
+            data-workload-key={workloadKey(workload)}
             attributes={{ CardContent: { className: 'workload-card-content' } }}
         >
             <div className="workload-badge hidden" style={{ backgroundColor: barColor }}></div>

desktop/src/ui/components/Workloads/WorkloadListView.tsx

@@ -63,9 +63,7 @@ function WorkloadListView({ filter, setFilter }: WorkloadListViewProps) {
         const filterKey = Object.entries(filter)
             .map(([_, value]) => `${value}`)
             .join('-')
-        const workloadsKey = currentWorkloads
-            .map(w => workloadKey(w.originatedFrom, w.id))
-            .join('-')
+        const workloadsKey = currentWorkloads.map(w => workloadKey(w)).join('-')
         return `${filterKey}-${workloadsKey}`
     }, [filter, currentWorkloads])
 
@@ -110,7 +108,7 @@ function WorkloadListView({ filter, setFilter }: WorkloadListViewProps) {
                                 data-workload-list-content
                             >
                                 {currentWorkloads.map(workload => {
-                                    const key = workloadKey(workload.originatedFrom, workload.id)
+                                    const key = workloadKey(workload)
                                     return (
                                         <Flipped key={key} flipId={key}>
                                             <div>

desktop/src/ui/components/Workloads/WorkloadNodeConnections.tsx

@@ -50,14 +50,6 @@ function connectionsEqual(a: readonly Connection[], b: readonly Connection[]): b
     return true
 }
 
-// Build a key that uniquely identifies a workload's DOM anchor. Workload ids are
-// a per-node proxy counter and collide across nodes, so we key on the
-// (origin, id) pair the same way `workloadKey` does. The null byte can't appear
-// in either value, so it is a safe separator.
-function anchorKey(origin: string, id: string): string {
-    return `${origin}\u0000${id}`
-}
-
 function computeConnections(svg: SVGSVGElement, workloads: readonly Workload[]): Connection[] {
     const svgRect = svg.getBoundingClientRect()
     const result: Connection[] = []
@@ -66,10 +58,9 @@ function computeConnections(svg: SVGSVGElement, workloads: readonly Workload[]):
     // frame. With N running jobs the old path issued 2 * N selector queries every
     // animation frame; this issues two querySelectorAll calls total.
     const workloadElByKey = new Map<string, Element>()
-    for (const el of document.querySelectorAll('[data-workload-id]')) {
-        const id = el.getAttribute('data-workload-id')
-        if (id === null) continue
-        workloadElByKey.set(anchorKey(el.getAttribute('data-workload-origin') ?? '', id), el)
+    for (const el of document.querySelectorAll('[data-workload-key]')) {
+        const key = el.getAttribute('data-workload-key')
+        if (key !== null) workloadElByKey.set(key, el)
     }
     const nodeElById = new Map<string, Element>()
     for (const el of document.querySelectorAll('[data-node-id]')) {
@@ -81,9 +72,7 @@ function computeConnections(svg: SVGSVGElement, workloads: readonly Workload[]):
         const executionNodeId = workloadExecutionNodeId(workload)
         if (!executionNodeId) continue
 
-        const workloadEl = workloadElByKey.get(
-            anchorKey(workload.originatedFrom ?? '', workload.id)
-        )
+        const workloadEl = workloadElByKey.get(workloadKey(workload))
         const nodeEl = nodeElById.get(executionNodeId)
         if (!workloadEl || !nodeEl) continue
 
@@ -126,7 +115,7 @@ function computeConnections(svg: SVGSVGElement, workloads: readonly Workload[]):
 
         const color = getWorkloadStateColor(workload.state)
         result.push({
-            workloadId: workloadKey(workload.originatedFrom, workload.id),
+            workloadId: workloadKey(workload),
             nodeId: executionNodeId,
             color: `${WORKLOAD_COLOR_MAP[color]}${workload.state === 'running' ? 'FF' : '88'}`,
             path,