From c584999346d0af63d6a0e3d8107c6fde4623a533 Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 16:02:58 +0000 Subject: [PATCH 1/7] perf(ui): localize slow input edge alerts Signed-off-by: streamkit-devin --- ui/src/components/TypedEdge.tsx | 122 +++++++- .../components/monitor.render-perf.test.tsx | 287 ++++++++++++++++++ ui/src/components/node/PinRow.tsx | 39 +-- ui/src/hooks/useEdgeAlertSubscription.test.ts | 234 -------------- ui/src/hooks/useEdgeAlertSubscription.ts | 200 ------------ ui/src/utils/pipelineGraph.test.ts | 20 ++ ui/src/utils/pipelineGraph.ts | 51 +++- ui/src/views/MonitorView.tsx | 23 +- 8 files changed, 488 insertions(+), 488 deletions(-) create mode 100644 ui/src/components/monitor.render-perf.test.tsx delete mode 100644 ui/src/hooks/useEdgeAlertSubscription.test.ts delete mode 100644 ui/src/hooks/useEdgeAlertSubscription.ts diff --git a/ui/src/components/TypedEdge.tsx b/ui/src/components/TypedEdge.tsx index 9ef7ab5c4..0cf8c22ee 100644 --- a/ui/src/components/TypedEdge.tsx +++ b/ui/src/components/TypedEdge.tsx @@ -2,15 +2,33 @@ // // SPDX-License-Identifier: MPL-2.0 -import { BaseEdge, EdgeLabelRenderer, getBezierPath, type EdgeProps } from '@xyflow/react'; +import { + BaseEdge, + EdgeLabelRenderer, + getBezierPath, + type Edge, + type EdgeProps, +} from '@xyflow/react'; +import { atom, type Atom } from 'jotai'; +import { useAtomValue } from 'jotai/react'; +import { selectAtom } from 'jotai/utils'; import React from 'react'; import { SKTooltip } from '@/components/Tooltip'; +import { nodeKey, nodeStateAtom } from '@/stores/sessionAtoms'; import type { PacketType } from '@/types/types'; +import { + describeSlowInputsFromConnections, + extractSlowTimeoutDetailsFromNodeState, + type MonitorEdgeAlertContext, + type SlowTimeoutDetails, +} from '@/utils/pipelineGraph'; +import { deepEqual } from '@/utils/deepEqual'; import { getPacketTypeColor } from '@/utils/packetTypes'; export type TypedEdgeData = { resolvedType?: PacketType; + monitorAlertContext?: MonitorEdgeAlertContext; alert?: { kind: string; severity: 'warning' | 'error'; @@ -23,6 +41,94 @@ export type TypedEdgeData = { }; type TypedEdgeAlert = NonNullable; +type SlowInputDetailsAtom = Atom; +type AlertEdge = Pick; + +const nullSlowInputDetailsAtom = atom(null); +const slowInputDetailsAtoms = new Map(); + +function getSlowInputDetailsAtom( + sessionId: string, + targetNode: string, + targetHandle: string +): SlowInputDetailsAtom { + const key = `${sessionId}\u0000${targetNode}\u0000${targetHandle}`; + const existing = slowInputDetailsAtoms.get(key); + if (existing) return existing; + + const selected = selectAtom( + nodeStateAtom(nodeKey(sessionId, targetNode)), + (state) => { + const details = extractSlowTimeoutDetailsFromNodeState(state); + if (!details || !details.slowPins.includes(targetHandle)) return null; + return details; + }, + deepEqual + ); + slowInputDetailsAtoms.set(key, selected); + return selected; +} + +function buildSlowInputTooltipLines( + edge: AlertEdge, + details: SlowTimeoutDetails, + connections: MonitorEdgeAlertContext['connections'] +): string[] { + const slowInputs = describeSlowInputsFromConnections(connections, edge.target, details.slowPins); + const lines: string[] = []; + if (slowInputs.length > 0) { + lines.push(`Slow inputs: ${slowInputs.join(', ')}`); + } else if (details.slowPins.length > 0) { + lines.push(`Slow pins: ${details.slowPins.join(', ')}`); + } + + lines.push(`This: ${edge.source}.${edge.sourceHandle ?? ''} → ${edge.targetHandle ?? ''}`); + + if (details.newlySlowPins.length > 0) { + lines.push(`Newly slow: ${details.newlySlowPins.join(', ')}`); + } + if (details.syncTimeoutMs != null) { + lines.push(`Timeout: ${details.syncTimeoutMs}ms`); + } + return lines; +} + +export function buildSlowInputAlert( + edge: AlertEdge, + details: SlowTimeoutDetails | null, + connections: MonitorEdgeAlertContext['connections'] +): TypedEdgeAlert | null { + if (!details) return null; + return { + kind: 'slow_input_timeout', + severity: 'warning', + tooltip: { + title: `${edge.target} degraded`, + lines: buildSlowInputTooltipLines(edge, details, connections), + }, + }; +} + +export function useSlowInputAlert( + edge: AlertEdge, + monitorAlertContext: MonitorEdgeAlertContext | undefined +): TypedEdgeAlert | null { + const detailsAtom = React.useMemo( + () => + monitorAlertContext + ? getSlowInputDetailsAtom( + monitorAlertContext.sessionId, + edge.target, + edge.targetHandle ?? '' + ) + : nullSlowInputDetailsAtom, + [edge.target, edge.targetHandle, monitorAlertContext] + ); + const details = useAtomValue(detailsAtom); + return monitorAlertContext + ? buildSlowInputAlert(edge, details, monitorAlertContext.connections) + : null; +} function getTypeColor(resolvedType: PacketType | undefined): string { return resolvedType ? getPacketTypeColor(resolvedType) : 'var(--sk-primary)'; @@ -84,6 +190,8 @@ const TypedEdge: React.FC = ({ targetPosition, style = {}, data, + source, + target, }) => { const [edgePath, labelX, labelY] = getBezierPath({ sourceX, @@ -96,7 +204,17 @@ const TypedEdge: React.FC = ({ const typedData = data as TypedEdgeData | undefined; const resolvedType = typedData?.resolvedType; - const alert = typedData?.alert; + const monitorAlertContext = typedData?.monitorAlertContext; + const dynamicAlert = useSlowInputAlert( + { + source, + sourceHandle: monitorAlertContext?.sourceHandle, + target, + targetHandle: monitorAlertContext?.targetHandle, + }, + monitorAlertContext + ); + const alert = dynamicAlert ?? typedData?.alert; const typeColor = getTypeColor(resolvedType); const alertColor = getAlertColor(alert); diff --git a/ui/src/components/monitor.render-perf.test.tsx b/ui/src/components/monitor.render-perf.test.tsx new file mode 100644 index 000000000..9489075c7 --- /dev/null +++ b/ui/src/components/monitor.render-perf.test.tsx @@ -0,0 +1,287 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import { act, render, renderHook, waitFor } from '@testing-library/react'; +import { ReactFlowProvider, useStoreApi, type Node } from '@xyflow/react'; +import React, { useEffect } from 'react'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { PinRow } from '@/components/node/PinRow'; +import { PinHandle } from '@/components/node/PinHandle'; +import { sessionStore, nodeKey, nodeStateAtom } from '@/stores/sessionAtoms'; +import type { Connection, NodeState, OutputPin } from '@/types/types'; + +import { buildSlowInputAlert, useSlowInputAlert } from './TypedEdge'; + +vi.mock('@/components/node/PinHandle', () => ({ + PinHandle: vi.fn(({ name, packetType }: { name: string; packetType: unknown }) => ( + {JSON.stringify(packetType)} + )), +})); + +const SESSION_ID = 'monitor-perf-session'; +const TARGET_NODE = 'mixer'; +const edgeContext = { + sessionId: SESSION_ID, + sourceHandle: 'audio', + targetHandle: 'audio_in', + connections: [ + { from_node: 'source', from_pin: 'audio', to_node: TARGET_NODE, to_pin: 'audio_in' }, + { from_node: 'source', from_pin: 'video', to_node: TARGET_NODE, to_pin: 'video_in' }, + ] satisfies Connection[], +}; + +const edge = { + source: 'source', + sourceHandle: 'audio', + target: TARGET_NODE, + targetHandle: 'audio_in', +}; + +function slowState( + slowPins: string[], + newlySlowPins: string[] = slowPins, + syncTimeoutMs = 500 +): NodeState { + return { + Degraded: { + reason: 'slow_input_timeout', + details: { + slow_pins: slowPins, + newly_slow_pins: newlySlowPins, + sync_timeout_ms: syncTimeoutMs, + }, + }, + }; +} + +function resetTargetState(): void { + sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, TARGET_NODE)), null); +} + +function AlertProbe({ + targetHandle, + onRender, +}: { + targetHandle: string; + onRender: (alert: ReturnType) => void; +}) { + const alert = useSlowInputAlert({ ...edge, targetHandle }, edgeContext); + onRender(alert); + return null; +} + +describe('Monitor edge alert render isolation', () => { + beforeEach(resetTargetState); + + it('updates only the exact target pin and ignores unrelated nodes', () => { + const audioRenders: ReturnType[] = []; + const videoRenders: ReturnType[] = []; + render( + <> + audioRenders.push(alert)} /> + videoRenders.push(alert)} /> + + ); + const initialAudioRenders = audioRenders.length; + const initialVideoRenders = videoRenders.length; + + act(() => { + sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, TARGET_NODE)), slowState(['audio_in'])); + }); + + expect(audioRenders.at(-1)).not.toBeNull(); + expect(videoRenders.length).toBe(initialVideoRenders); + + act(() => { + sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'unrelated')), slowState(['audio_in'])); + }); + + expect(audioRenders.length).toBeGreaterThan(initialAudioRenders); + expect(videoRenders.length).toBe(initialVideoRenders); + }); + + it('warns on matching pins, ignores nonmatching pins, and clears on recovery', () => { + const { result } = renderHook(() => useSlowInputAlert(edge, edgeContext)); + + expect(result.current).toBeNull(); + act(() => { + sessionStore.set( + nodeStateAtom(nodeKey(SESSION_ID, TARGET_NODE)), + slowState(['video_in'], ['video_in']) + ); + }); + expect(result.current).toBeNull(); + + act(() => { + sessionStore.set( + nodeStateAtom(nodeKey(SESSION_ID, TARGET_NODE)), + slowState(['audio_in'], ['audio_in']) + ); + }); + expect(result.current?.kind).toBe('slow_input_timeout'); + expect(result.current?.severity).toBe('warning'); + + act(() => { + sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, TARGET_NODE)), 'Running'); + }); + expect(result.current).toBeNull(); + }); + + it('keeps the existing tooltip title and source detail lines', () => { + const alert = buildSlowInputAlert( + edge, + { + slowPins: ['audio_in', 'missing_in'], + newlySlowPins: ['audio_in'], + syncTimeoutMs: 750, + }, + edgeContext.connections + ); + + expect(alert).toEqual({ + kind: 'slow_input_timeout', + severity: 'warning', + tooltip: { + title: 'mixer degraded', + lines: [ + 'Slow inputs: source.audio → audio_in', + 'This: source.audio → audio_in', + 'Newly slow: audio_in', + 'Timeout: 750ms', + ], + }, + }); + }); +}); + +let flowStore: ReturnType | null = null; + +function StoreProbe() { + const store = useStoreApi(); + useEffect(() => { + flowStore = store; + }, [store]); + return null; +} + +function FlowHarness({ children }: { children: React.ReactNode }) { + return ( + + + {children} + + ); +} + +const passthroughPin: OutputPin = { + name: 'out', + produces_type: 'Passthrough', + cardinality: 'One', +}; + +const sourceNode: Node = { + id: 'source', + type: 'test', + position: { x: 0, y: 0 }, + data: { inputs: [], outputs: [{ name: 'out', produces_type: 'Text', cardinality: 'One' }] }, +}; + +const passthroughNode: Node = { + id: 'passthrough', + type: 'test', + position: { x: 0, y: 0 }, + data: { inputs: [{ name: 'in', accepts_types: ['Any'], cardinality: 'One' }], outputs: [] }, +}; + +describe('PinRow render isolation', () => { + beforeEach(() => { + flowStore = null; + }); + + it('ignores edge-data-only updates while following relevant topology type changes', async () => { + const pinHandleRenders = vi.fn(); + vi.mocked(PinHandle).mockImplementation(({ name, packetType }) => { + pinHandleRenders({ name, packetType }); + return {JSON.stringify(packetType)}; + }); + + render( + + + + ); + await waitFor(() => expect(flowStore).not.toBeNull()); + + act(() => { + flowStore!.getState().setNodes([sourceNode, passthroughNode]); + flowStore!.getState().setEdges([ + { + id: 'source-out-passthrough-in', + source: 'source', + sourceHandle: 'out', + target: 'passthrough', + targetHandle: 'in', + data: { alert: { kind: 'slow_input_timeout' } }, + }, + ]); + }); + await waitFor(() => expect(pinHandleRenders).toHaveBeenCalled()); + pinHandleRenders.mockClear(); + + act(() => { + flowStore!.getState().setEdges([ + { + id: 'source-out-passthrough-in', + source: 'source', + sourceHandle: 'out', + target: 'passthrough', + targetHandle: 'in', + data: { alert: { kind: 'other' } }, + }, + ]); + }); + expect(pinHandleRenders).not.toHaveBeenCalled(); + + act(() => { + flowStore!.getState().setNodes([ + { + ...sourceNode, + data: { + ...sourceNode.data, + outputs: [{ name: 'out', produces_type: 'Binary', cardinality: 'One' }], + }, + }, + passthroughNode, + ]); + }); + await waitFor(() => + expect(pinHandleRenders).toHaveBeenCalledWith( + expect.objectContaining({ packetType: 'Binary' }) + ) + ); + }); + + it('returns null for rows without passthrough outputs', async () => { + const pinHandleRenders = vi.fn(); + vi.mocked(PinHandle).mockImplementation(({ name, packetType }) => { + pinHandleRenders({ name, packetType }); + return {JSON.stringify(packetType)}; + }); + + render( + + + + ); + await waitFor(() => + expect(pinHandleRenders).toHaveBeenCalledWith(expect.objectContaining({ packetType: 'Text' })) + ); + }); +}); diff --git a/ui/src/components/node/PinRow.tsx b/ui/src/components/node/PinRow.tsx index e2c58c4ba..76b257891 100644 --- a/ui/src/components/node/PinRow.tsx +++ b/ui/src/components/node/PinRow.tsx @@ -2,8 +2,8 @@ // // SPDX-License-Identifier: MPL-2.0 -import { Position, useUpdateNodeInternals, useStore } from '@xyflow/react'; -import React, { useEffect, useMemo } from 'react'; +import { Position, useUpdateNodeInternals, useStore, type ReactFlowState } from '@xyflow/react'; +import React, { useEffect } from 'react'; import type { InputPin, OutputPin, PacketType } from '@/types/types'; @@ -65,8 +65,15 @@ type PinRowProps = { export const PinRow: React.FC = ({ nodeId, side, pins, isInput, totalPins }) => { const update = useUpdateNodeInternals(); - const edges = useStore((state) => state.edges); - const nodes = useStore((state) => state.nodeLookup); + const hasPassthroughOutput = + !isInput && pins.some((pin) => (pin as OutputPin).produces_type === 'Passthrough'); + const resolvedPassthroughType = useStore( + React.useCallback( + (state: ReactFlowState): PacketType | null => + hasPassthroughOutput ? findUpstreamOutputType(nodeId, state.nodeLookup, state.edges) : null, + [hasPassthroughOutput, nodeId] + ) + ); useEffect(() => { update(nodeId); @@ -83,26 +90,6 @@ export const PinRow: React.FC = ({ nodeId, side, pins, isInput, tot const total = totalPins ?? pins.length; - // Resolve Passthrough output pin types based on connections - const resolvedOutputTypes = useMemo(() => { - if (isInput) return new Map(); - - const resolved = new Map(); - for (const pin of pins) { - const outputPin = pin as OutputPin; - - // Skip non-Passthrough pins - they have explicit types - if (outputPin.produces_type !== 'Passthrough') continue; - - // Resolve the type by finding upstream connection - const resolvedType = findUpstreamOutputType(nodeId, nodes, edges); - if (resolvedType) { - resolved.set(outputPin.name, resolvedType); - } - } - return resolved; - }, [isInput, pins, edges, nodes, nodeId]); - return ( <> {pins.map((p, i: number) => { @@ -113,7 +100,9 @@ export const PinRow: React.FC = ({ nodeId, side, pins, isInput, tot packetType = ((p as InputPin).accepts_types?.[0] ?? 'Any') as PacketType; } else { const outputPin = p as OutputPin; - packetType = resolvedOutputTypes.get(name) || outputPin.produces_type; + packetType = + (outputPin.produces_type === 'Passthrough' ? resolvedPassthroughType : null) || + outputPin.produces_type; } const cardinality = p.cardinality; diff --git a/ui/src/hooks/useEdgeAlertSubscription.test.ts b/ui/src/hooks/useEdgeAlertSubscription.test.ts deleted file mode 100644 index 655ee3a31..000000000 --- a/ui/src/hooks/useEdgeAlertSubscription.test.ts +++ /dev/null @@ -1,234 +0,0 @@ -// SPDX-FileCopyrightText: © 2025 StreamKit Contributors -// -// SPDX-License-Identifier: MPL-2.0 - -import { renderHook, act } from '@testing-library/react'; -import type { Edge } from '@xyflow/react'; -import React from 'react'; -import { describe, it, expect, afterEach, vi } from 'vitest'; - -import { sessionStore, nodeStateAtom, nodeKey, clearSessionAtoms } from '@/stores/sessionAtoms'; -import { useSessionStore } from '@/stores/sessionStore'; -import type { Pipeline, NodeState } from '@/types/types'; -import { isRecord } from '@/utils/pipelineGraph'; - -import type { UseEdgeAlertSubscriptionOptions } from './useEdgeAlertSubscription'; -import { useEdgeAlertSubscription } from './useEdgeAlertSubscription'; - -const SESSION_ID = 'test-session-edge-alerts'; - -function makePipeline(nodes: Record): Pipeline { - const mapped: Pipeline['nodes'] = {}; - for (const [id, n] of Object.entries(nodes)) { - mapped[id] = { kind: n.kind, params: {}, state: n.state ?? null }; - } - return { - name: null, - description: null, - mode: 'dynamic', - client: null, - nodes: mapped, - connections: [{ from_node: 'source', from_pin: 'out', to_node: 'mixer', to_pin: 'audio_in' }], - }; -} - -function makeEdges(): Edge[] { - return [ - { - id: 'source-mixer', - source: 'source', - sourceHandle: 'out', - target: 'mixer', - targetHandle: 'audio_in', - }, - ]; -} - -function makeOptions( - overrides: Partial & { - pipeline?: Pipeline; - edges?: Edge[]; - } = {} -): { - options: UseEdgeAlertSubscriptionOptions; - getEdges: () => Edge[]; -} { - const pipeline = - overrides.pipeline ?? - makePipeline({ - source: { kind: 'core::passthrough' }, - mixer: { kind: 'core::mixer' }, - }); - let edges = overrides.edges ?? makeEdges(); - const setEdges: React.Dispatch> = (updater) => { - edges = typeof updater === 'function' ? updater(edges) : updater; - }; - const pipelineRef = { current: pipeline }; - - return { - options: { - selectedSessionId: overrides.selectedSessionId ?? SESSION_ID, - setEdges: overrides.setEdges ?? setEdges, - pipelineRef: pipelineRef as React.RefObject, - topoKey: overrides.topoKey ?? 'topo-1', - }, - getEdges: () => edges, - }; -} - -afterEach(() => { - clearSessionAtoms(SESSION_ID); - useSessionStore.getState().clearSession(SESSION_ID); -}); - -describe('useEdgeAlertSubscription', () => { - it('returns topoEffectRanRef that gates edge patching', async () => { - const { options, getEdges } = makeOptions(); - - const { result } = renderHook(() => useEdgeAlertSubscription(options)); - - expect(result.current.topoEffectRanRef.current).toBe(false); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), { - Degraded: { - reason: 'slow_input_timeout', - details: { slow_pins: ['audio_in'], newly_slow_pins: ['audio_in'], sync_timeout_ms: 100 }, - }, - }); - }); - - const edgesAfterGated = getEdges(); - const alert = isRecord(edgesAfterGated[0].data) ? edgesAfterGated[0].data['alert'] : undefined; - expect(alert).toBeUndefined(); - }); - - it('patches edges with alert data when a node enters slow_input_timeout', async () => { - const { options, getEdges } = makeOptions(); - - const { result } = renderHook(() => useEdgeAlertSubscription(options)); - - act(() => { - result.current.topoEffectRanRef.current = true; - }); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), { - Degraded: { - reason: 'slow_input_timeout', - details: { slow_pins: ['audio_in'], newly_slow_pins: ['audio_in'], sync_timeout_ms: 100 }, - }, - }); - }); - - const patched = getEdges(); - const alertData = isRecord(patched[0].data) ? patched[0].data['alert'] : undefined; - expect(alertData).toBeDefined(); - expect(isRecord(alertData) && alertData['kind']).toBe('slow_input_timeout'); - }); - - it('clears edge alert when node recovers from degraded state', async () => { - const { options, getEdges } = makeOptions(); - - const { result } = renderHook(() => useEdgeAlertSubscription(options)); - - act(() => { - result.current.topoEffectRanRef.current = true; - }); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), { - Degraded: { - reason: 'slow_input_timeout', - details: { slow_pins: ['audio_in'], newly_slow_pins: [], sync_timeout_ms: 100 }, - }, - }); - }); - - const degraded = getEdges(); - const alertBefore = isRecord(degraded[0].data) ? degraded[0].data['alert'] : undefined; - expect(alertBefore).toBeDefined(); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), 'Running'); - }); - - const recovered = getEdges(); - const alertAfter = isRecord(recovered[0].data) ? recovered[0].data['alert'] : undefined; - expect(alertAfter).toBeUndefined(); - }); - - it('does not patch edges for a non-matching target handle', async () => { - const edges: Edge[] = [ - { - id: 'source-mixer', - source: 'source', - sourceHandle: 'out', - target: 'mixer', - targetHandle: 'video_in', - }, - ]; - const { options, getEdges } = makeOptions({ edges }); - - const { result } = renderHook(() => useEdgeAlertSubscription(options)); - - act(() => { - result.current.topoEffectRanRef.current = true; - }); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), { - Degraded: { - reason: 'slow_input_timeout', - details: { slow_pins: ['audio_in'], newly_slow_pins: ['audio_in'], sync_timeout_ms: 100 }, - }, - }); - }); - - const patched = getEdges(); - const alert = isRecord(patched[0].data) ? patched[0].data['alert'] : undefined; - expect(alert).toBeUndefined(); - }); - - it('resets topoEffectRanRef when selectedSessionId changes', () => { - const { options } = makeOptions(); - - const { result, rerender } = renderHook( - (props: UseEdgeAlertSubscriptionOptions) => useEdgeAlertSubscription(props), - { initialProps: options } - ); - - act(() => { - result.current.topoEffectRanRef.current = true; - }); - expect(result.current.topoEffectRanRef.current).toBe(true); - - const newOptions = { ...options, selectedSessionId: 'other-session' }; - rerender(newOptions); - - expect(result.current.topoEffectRanRef.current).toBe(false); - }); - - it('uses React.startTransition for edge updates', async () => { - const startTransitionSpy = vi.spyOn(React, 'startTransition'); - const { options } = makeOptions(); - - const { result } = renderHook(() => useEdgeAlertSubscription(options)); - - act(() => { - result.current.topoEffectRanRef.current = true; - }); - - await act(async () => { - sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'mixer')), { - Degraded: { - reason: 'slow_input_timeout', - details: { slow_pins: ['audio_in'], newly_slow_pins: [], sync_timeout_ms: 100 }, - }, - }); - }); - - expect(startTransitionSpy).toHaveBeenCalled(); - startTransitionSpy.mockRestore(); - }); -}); diff --git a/ui/src/hooks/useEdgeAlertSubscription.ts b/ui/src/hooks/useEdgeAlertSubscription.ts deleted file mode 100644 index 0137be287..000000000 --- a/ui/src/hooks/useEdgeAlertSubscription.ts +++ /dev/null @@ -1,200 +0,0 @@ -// SPDX-FileCopyrightText: © 2025 StreamKit Contributors -// -// SPDX-License-Identifier: MPL-2.0 - -/** - * Hook that patches ReactFlow edge alert metadata (slow-input-timeout - * warnings) by subscribing directly to per-node Jotai state atoms. - * - * Node components read their `state` from per-node atoms (via - * {@link useNodeStateFromAtom}); this hook only patches edge `data.alert` - * so that warning badges appear on affected edges. - */ - -import type { Edge } from '@xyflow/react'; -import React, { useEffect, useRef } from 'react'; - -import { sessionStore, nodeStateAtom, nodeKey } from '@/stores/sessionAtoms'; -import type { NodeState, Pipeline } from '@/types/types'; -import { - isRecord, - extractSlowTimeoutDetailsFromNodeState, - describeSlowInputs, - type SlowTimeoutDetails, -} from '@/utils/pipelineGraph'; - -function buildSlowInputTooltipLines( - edge: Edge, - details: SlowTimeoutDetails | undefined, - pipeline: Pipeline -): string[] { - const slowPins = details?.slowPins ?? []; - const slowInputs = describeSlowInputs(pipeline, edge.target, slowPins); - - const lines: string[] = []; - if (slowInputs.length > 0) { - lines.push(`Slow inputs: ${slowInputs.join(', ')}`); - } else if (slowPins.length > 0) { - lines.push(`Slow pins: ${slowPins.join(', ')}`); - } - - lines.push(`This: ${edge.source}.${edge.sourceHandle ?? ''} → ${edge.targetHandle ?? ''}`); - - if (details?.newlySlowPins && details.newlySlowPins.length > 0) { - lines.push(`Newly slow: ${details.newlySlowPins.join(', ')}`); - } - if (details?.syncTimeoutMs != null) { - lines.push(`Timeout: ${details.syncTimeoutMs}ms`); - } - return lines; -} - -function buildEdgeAlert( - edge: Edge, - slowPinsByNode: Map>, - slowDetailsByNode: Map, - pipeline: Pipeline -): Record | null { - const shouldWarn = slowPinsByNode.get(edge.target)?.has(edge.targetHandle ?? '') ?? false; - if (!shouldWarn) return null; - - const details = slowDetailsByNode.get(edge.target); - return { - kind: 'slow_input_timeout', - severity: 'warning', - tooltip: { - title: `${edge.target} degraded`, - lines: buildSlowInputTooltipLines(edge, details, pipeline), - }, - }; -} - -function collectSlowPinData( - pipeline: Pipeline, - nodeStates: Record -): { - slowPinsByNode: Map>; - slowDetailsByNode: Map; -} { - const slowPinsByNode = new Map>(); - const slowDetailsByNode = new Map(); - for (const [nodeId, apiNode] of Object.entries(pipeline.nodes)) { - const st = nodeStates[nodeId] ?? apiNode.state ?? null; - const details = extractSlowTimeoutDetailsFromNodeState(st); - const slowPins = details?.slowPins ?? []; - if (slowPins.length > 0) { - slowPinsByNode.set(nodeId, new Set(slowPins)); - } - if (details) { - slowDetailsByNode.set(nodeId, details); - } - } - return { slowPinsByNode, slowDetailsByNode }; -} - -export interface UseEdgeAlertSubscriptionOptions { - selectedSessionId: string | null; - setEdges: React.Dispatch>; - pipelineRef: React.RefObject; - topoKey: string; -} - -export interface UseEdgeAlertSubscriptionReturn { - topoEffectRanRef: React.MutableRefObject; -} - -export function useEdgeAlertSubscription({ - selectedSessionId, - setEdges, - pipelineRef, - topoKey, -}: UseEdgeAlertSubscriptionOptions): UseEdgeAlertSubscriptionReturn { - const topoEffectRanRef = useRef(false); - - useEffect(() => { - if (!selectedSessionId) return; - - // Must run before MonitorView's topology effect, which re-sets the ref to true. - topoEffectRanRef.current = false; - - const pipeline = pipelineRef.current; - const nodeIds = pipeline ? Object.keys(pipeline.nodes) : []; - - const applyPatch = () => { - const currentPipeline = pipelineRef.current; - if (!currentPipeline) return; - - const nodeStates: Record = {}; - for (const nodeId of Object.keys(currentPipeline.nodes)) { - const state = sessionStore.get(nodeStateAtom(nodeKey(selectedSessionId, nodeId))); - if (state) nodeStates[nodeId] = state; - } - - React.startTransition(() => { - const { slowPinsByNode, slowDetailsByNode } = collectSlowPinData( - currentPipeline, - nodeStates - ); - - setEdges((prev) => { - let changed = false; - - const next = prev.map((edge) => { - const targetPin = edge.targetHandle ?? ''; - const shouldWarn = slowPinsByNode.get(edge.target)?.has(targetPin) ?? false; - const currentAlert = isRecord(edge.data) ? edge.data['alert'] : undefined; - const currentAlertKind = - isRecord(currentAlert) && typeof currentAlert['kind'] === 'string' - ? currentAlert['kind'] - : null; - const isCurrentlyWarned = currentAlertKind === 'slow_input_timeout'; - - if (shouldWarn === isCurrentlyWarned) return edge; - - changed = true; - const nextData: Record = { ...(edge.data || {}) }; - - if (shouldWarn) { - nextData.alert = buildEdgeAlert( - edge, - slowPinsByNode, - slowDetailsByNode, - currentPipeline - ); - } else if (isCurrentlyWarned) { - delete nextData.alert; - } - - return { ...edge, data: nextData }; - }); - - return changed ? next : prev; - }); - }); - }; - - let disposed = false; - let pendingFlush = false; - const onAtomChange = () => { - if (!topoEffectRanRef.current) return; - if (pendingFlush) return; - pendingFlush = true; - queueMicrotask(() => { - pendingFlush = false; - if (disposed) return; - applyPatch(); - }); - }; - - const unsubs = nodeIds.map((id) => - sessionStore.sub(nodeStateAtom(nodeKey(selectedSessionId, id)), onAtomChange) - ); - - return () => { - disposed = true; - unsubs.forEach((u) => u()); - }; - }, [selectedSessionId, setEdges, pipelineRef, topoKey]); - - return { topoEffectRanRef }; -} diff --git a/ui/src/utils/pipelineGraph.test.ts b/ui/src/utils/pipelineGraph.test.ts index 984948ae2..b5855b931 100644 --- a/ui/src/utils/pipelineGraph.test.ts +++ b/ui/src/utils/pipelineGraph.test.ts @@ -352,6 +352,26 @@ describe('buildEdgesFromConnections', () => { }); }); + it('attaches monitor alert context without changing design edge data', () => { + const nodes = [makeRFNode('a', [], [outPinOne('out')]), makeRFNode('b', [pinOne('in')], [])]; + const connections: Connection[] = [ + { from_node: 'a', from_pin: 'out', to_node: 'b', to_pin: 'in' }, + ]; + const monitorContext = { sessionId: 'session-1', connections }; + + const designEdges = buildEdgesFromConnections(connections, nodes); + const monitorEdges = buildEdgesFromConnections(connections, nodes, monitorContext); + + expect(designEdges[0]?.data).toBeUndefined(); + expect(monitorEdges[0]?.data).toEqual({ + monitorAlertContext: { + ...monitorContext, + sourceHandle: 'out', + targetHandle: 'in', + }, + }); + }); + it('produces stable ids regardless of input ordering', () => { const nodes = [ makeRFNode('a', [], [outPinOne('out')]), diff --git a/ui/src/utils/pipelineGraph.ts b/ui/src/utils/pipelineGraph.ts index 934c767c9..42a792204 100644 --- a/ui/src/utils/pipelineGraph.ts +++ b/ui/src/utils/pipelineGraph.ts @@ -23,6 +23,13 @@ export type SlowTimeoutDetails = { syncTimeoutMs: number | null; }; +export type MonitorEdgeAlertContext = { + sessionId: string; + connections: Connection[]; + sourceHandle?: string; + targetHandle?: string; +}; + export const isRecord = (value: unknown): value is Record => value !== null && value !== undefined && typeof value === 'object' && !Array.isArray(value); @@ -51,15 +58,15 @@ export const extractSlowTimeoutDetailsFromNodeState = ( return { slowPins, newlySlowPins, syncTimeoutMs }; }; -export const describeSlowInputs = ( - pipeline: Pipeline, +export const describeSlowInputsFromConnections = ( + connections: Connection[], nodeId: string, slowPins: string[] ): string[] => { if (slowPins.length === 0) return []; const slowPinSet = new Set(slowPins); - const sources = pipeline.connections + const sources = connections .filter((c) => c.to_node === nodeId && slowPinSet.has(c.to_pin)) .map((c) => `${c.from_node}.${c.from_pin} → ${c.to_pin}`); @@ -67,6 +74,12 @@ export const describeSlowInputs = ( return sources; }; +export const describeSlowInputs = ( + pipeline: Pipeline, + nodeId: string, + slowPins: string[] +): string[] => describeSlowInputsFromConnections(pipeline.connections, nodeId, slowPins); + // Edge connection validation const isValidEdgeConnection = (conn: Connection, nodeMap: Map): boolean => { @@ -91,18 +104,34 @@ const isValidEdgeConnection = (conn: Connection, nodeMap: Map): return hasSourcePin && hasTargetPin; }; -export const buildEdgesFromConnections = (connections: Connection[], nodes: RFNode[]): Edge[] => { +export const buildEdgesFromConnections = ( + connections: Connection[], + nodes: RFNode[], + monitorAlertContext?: MonitorEdgeAlertContext +): Edge[] => { const nodeMap = new Map(nodes.map((n) => [n.id, n])); return connections .filter((conn) => isValidEdgeConnection(conn, nodeMap)) - .map((conn) => ({ - id: `${conn.from_node}_${conn.from_pin}-${conn.to_node}_${conn.to_pin}`, - source: conn.from_node, - sourceHandle: conn.from_pin, - target: conn.to_node, - targetHandle: conn.to_pin, - })); + .map((conn) => { + const edge: Edge = { + id: `${conn.from_node}_${conn.from_pin}-${conn.to_node}_${conn.to_pin}`, + source: conn.from_node, + sourceHandle: conn.from_pin, + target: conn.to_node, + targetHandle: conn.to_pin, + }; + if (monitorAlertContext) { + edge.data = { + monitorAlertContext: { + ...monitorAlertContext, + sourceHandle: conn.from_pin, + targetHandle: conn.to_pin, + }, + }; + } + return edge; + }); }; // YAML generation diff --git a/ui/src/views/MonitorView.tsx b/ui/src/views/MonitorView.tsx index 02a05b47f..d2f3b3ffa 100644 --- a/ui/src/views/MonitorView.tsx +++ b/ui/src/views/MonitorView.tsx @@ -42,7 +42,6 @@ import { DnDProvider, useDnD } from '@/context/DnDContext'; import { useToast } from '@/context/ToastContext'; import { useAutoLayout } from '@/hooks/useAutoLayout'; import { useContextMenu } from '@/hooks/useContextMenu'; -import { useEdgeAlertSubscription } from '@/hooks/useEdgeAlertSubscription'; import { useMonitorPreview } from '@/hooks/useMonitorPreview'; import { useReactFlowCommon } from '@/hooks/useReactFlowCommon'; import { useResolvedColorMode } from '@/hooks/useResolvedColorMode'; @@ -398,11 +397,6 @@ const MonitorViewContent: React.FC = () => { handleStopPreview, } = useMonitorPreview(selectedSessionId); - const pipelineRef = useRef(pipeline); - useEffect(() => { - pipelineRef.current = pipeline; - }, [pipeline]); - useEffect(() => { const drafts = draftNodesRef.current; if (drafts.size === 0) return; @@ -524,13 +518,6 @@ const MonitorViewContent: React.FC = () => { setNeedsFit, ]); - const { topoEffectRanRef } = useEdgeAlertSubscription({ - selectedSessionId, - setEdges, - pipelineRef, - topoKey, - }); - const sessionSeenInListRef = useRef(false); useEffect(() => { if (selectedSession) { @@ -1069,7 +1056,13 @@ const MonitorViewContent: React.FC = () => { newNodes.push(node); } - const newEdges = buildEdgesFromConnections(pipeline?.connections ?? [], newNodes); + const newEdges = buildEdgesFromConnections( + pipeline?.connections ?? [], + newNodes, + selectedSessionId && pipeline + ? { sessionId: selectedSessionId, connections: pipeline.connections } + : undefined + ); for (const n of newNodes) { if (prevSelected.has(n.id)) n.selected = true; @@ -1079,7 +1072,6 @@ const MonitorViewContent: React.FC = () => { React.startTransition(() => { setNodes((prev) => (prev.length === 0 && newNodes.length === 0 ? prev : newNodes)); setEdges((prev) => (prev.length === 0 && newEdges.length === 0 ? prev : newEdges)); - topoEffectRanRef.current = true; }); const yamlString = pipeline ? generatePipelineYaml(pipeline, orderedNames) : ''; @@ -1101,7 +1093,6 @@ const MonitorViewContent: React.FC = () => { stableOnConfigChange, setNodes, setEdges, - topoEffectRanRef, ]); // Keep YAML in sync with live param overrides; runs only on param changes. From 876c44b1d63190b3a1844c731d4ba4b187a5074c Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 16:04:43 +0000 Subject: [PATCH 2/7] perf(ui): scope monitor alert atom lifetime Signed-off-by: streamkit-devin --- ui/src/components/TypedEdge.tsx | 54 +++++++------------ .../components/monitor.render-perf.test.tsx | 9 ++-- ui/src/utils/pipelineGraph.test.ts | 9 ++-- ui/src/utils/pipelineGraph.ts | 10 +--- 4 files changed, 27 insertions(+), 55 deletions(-) diff --git a/ui/src/components/TypedEdge.tsx b/ui/src/components/TypedEdge.tsx index 0cf8c22ee..4380dc4ac 100644 --- a/ui/src/components/TypedEdge.tsx +++ b/ui/src/components/TypedEdge.tsx @@ -45,29 +45,6 @@ type SlowInputDetailsAtom = Atom; type AlertEdge = Pick; const nullSlowInputDetailsAtom = atom(null); -const slowInputDetailsAtoms = new Map(); - -function getSlowInputDetailsAtom( - sessionId: string, - targetNode: string, - targetHandle: string -): SlowInputDetailsAtom { - const key = `${sessionId}\u0000${targetNode}\u0000${targetHandle}`; - const existing = slowInputDetailsAtoms.get(key); - if (existing) return existing; - - const selected = selectAtom( - nodeStateAtom(nodeKey(sessionId, targetNode)), - (state) => { - const details = extractSlowTimeoutDetailsFromNodeState(state); - if (!details || !details.slowPins.includes(targetHandle)) return null; - return details; - }, - deepEqual - ); - slowInputDetailsAtoms.set(key, selected); - return selected; -} function buildSlowInputTooltipLines( edge: AlertEdge, @@ -113,17 +90,20 @@ export function useSlowInputAlert( edge: AlertEdge, monitorAlertContext: MonitorEdgeAlertContext | undefined ): TypedEdgeAlert | null { - const detailsAtom = React.useMemo( - () => - monitorAlertContext - ? getSlowInputDetailsAtom( - monitorAlertContext.sessionId, - edge.target, - edge.targetHandle ?? '' - ) - : nullSlowInputDetailsAtom, - [edge.target, edge.targetHandle, monitorAlertContext] - ); + const sessionId = monitorAlertContext?.sessionId; + const targetHandle = edge.targetHandle ?? ''; + const detailsAtom = React.useMemo(() => { + if (!sessionId) return nullSlowInputDetailsAtom; + return selectAtom( + nodeStateAtom(nodeKey(sessionId, edge.target)), + (state) => { + const details = extractSlowTimeoutDetailsFromNodeState(state); + if (!details || !details.slowPins.includes(targetHandle)) return null; + return details; + }, + deepEqual + ); + }, [edge.target, sessionId, targetHandle]); const details = useAtomValue(detailsAtom); return monitorAlertContext ? buildSlowInputAlert(edge, details, monitorAlertContext.connections) @@ -192,6 +172,8 @@ const TypedEdge: React.FC = ({ data, source, target, + sourceHandleId, + targetHandleId, }) => { const [edgePath, labelX, labelY] = getBezierPath({ sourceX, @@ -208,9 +190,9 @@ const TypedEdge: React.FC = ({ const dynamicAlert = useSlowInputAlert( { source, - sourceHandle: monitorAlertContext?.sourceHandle, + sourceHandle: sourceHandleId, target, - targetHandle: monitorAlertContext?.targetHandle, + targetHandle: targetHandleId, }, monitorAlertContext ); diff --git a/ui/src/components/monitor.render-perf.test.tsx b/ui/src/components/monitor.render-perf.test.tsx index 9489075c7..c602dc654 100644 --- a/ui/src/components/monitor.render-perf.test.tsx +++ b/ui/src/components/monitor.render-perf.test.tsx @@ -24,8 +24,6 @@ const SESSION_ID = 'monitor-perf-session'; const TARGET_NODE = 'mixer'; const edgeContext = { sessionId: SESSION_ID, - sourceHandle: 'audio', - targetHandle: 'audio_in', connections: [ { from_node: 'source', from_pin: 'audio', to_node: TARGET_NODE, to_pin: 'audio_in' }, { from_node: 'source', from_pin: 'video', to_node: TARGET_NODE, to_pin: 'video_in' }, @@ -84,7 +82,6 @@ describe('Monitor edge alert render isolation', () => { videoRenders.push(alert)} /> ); - const initialAudioRenders = audioRenders.length; const initialVideoRenders = videoRenders.length; act(() => { @@ -93,13 +90,15 @@ describe('Monitor edge alert render isolation', () => { expect(audioRenders.at(-1)).not.toBeNull(); expect(videoRenders.length).toBe(initialVideoRenders); + const audioRendersAfterMatchingUpdate = audioRenders.length; + const videoRendersAfterMatchingUpdate = videoRenders.length; act(() => { sessionStore.set(nodeStateAtom(nodeKey(SESSION_ID, 'unrelated')), slowState(['audio_in'])); }); - expect(audioRenders.length).toBeGreaterThan(initialAudioRenders); - expect(videoRenders.length).toBe(initialVideoRenders); + expect(audioRenders.length).toBe(audioRendersAfterMatchingUpdate); + expect(videoRenders.length).toBe(videoRendersAfterMatchingUpdate); }); it('warns on matching pins, ignores nonmatching pins, and clears on recovery', () => { diff --git a/ui/src/utils/pipelineGraph.test.ts b/ui/src/utils/pipelineGraph.test.ts index b5855b931..12d978f25 100644 --- a/ui/src/utils/pipelineGraph.test.ts +++ b/ui/src/utils/pipelineGraph.test.ts @@ -364,12 +364,11 @@ describe('buildEdgesFromConnections', () => { expect(designEdges[0]?.data).toBeUndefined(); expect(monitorEdges[0]?.data).toEqual({ - monitorAlertContext: { - ...monitorContext, - sourceHandle: 'out', - targetHandle: 'in', - }, + monitorAlertContext: monitorContext, }); + expect((monitorEdges[0]?.data as { monitorAlertContext?: unknown })?.monitorAlertContext).toBe( + monitorContext + ); }); it('produces stable ids regardless of input ordering', () => { diff --git a/ui/src/utils/pipelineGraph.ts b/ui/src/utils/pipelineGraph.ts index 42a792204..1c7e1f0e2 100644 --- a/ui/src/utils/pipelineGraph.ts +++ b/ui/src/utils/pipelineGraph.ts @@ -26,8 +26,6 @@ export type SlowTimeoutDetails = { export type MonitorEdgeAlertContext = { sessionId: string; connections: Connection[]; - sourceHandle?: string; - targetHandle?: string; }; export const isRecord = (value: unknown): value is Record => @@ -122,13 +120,7 @@ export const buildEdgesFromConnections = ( targetHandle: conn.to_pin, }; if (monitorAlertContext) { - edge.data = { - monitorAlertContext: { - ...monitorAlertContext, - sourceHandle: conn.from_pin, - targetHandle: conn.to_pin, - }, - }; + edge.data = { monitorAlertContext }; } return edge; }); From 08c2dec053e3f34542b8a42c35974947794a82c8 Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 16:19:06 +0000 Subject: [PATCH 3/7] fix(ui): refresh monitor topology per session Signed-off-by: streamkit-devin --- ui/src/utils/monitorTopology.test.ts | 25 +++++++++++++++++++++++++ ui/src/utils/monitorTopology.ts | 8 ++++++++ ui/src/views/MonitorView.tsx | 6 ++++-- 3 files changed, 37 insertions(+), 2 deletions(-) create mode 100644 ui/src/utils/monitorTopology.test.ts create mode 100644 ui/src/utils/monitorTopology.ts diff --git a/ui/src/utils/monitorTopology.test.ts b/ui/src/utils/monitorTopology.test.ts new file mode 100644 index 000000000..cd193619e --- /dev/null +++ b/ui/src/utils/monitorTopology.test.ts @@ -0,0 +1,25 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import { describe, expect, it } from 'vitest'; + +import { buildMonitorTopologyKey } from './monitorTopology'; + +describe('buildMonitorTopologyKey', () => { + it('distinguishes identical topology fingerprints across sessions', () => { + const topologyFingerprint = JSON.stringify([ + ['source:audio', 'sink:audio'], + ['source:out>sink:in'], + ['source'], + [], + ]); + + const firstSessionKey = buildMonitorTopologyKey('session-1', topologyFingerprint); + const secondSessionKey = buildMonitorTopologyKey('session-2', topologyFingerprint); + + expect(firstSessionKey).not.toBe(secondSessionKey); + expect(JSON.parse(firstSessionKey)[0]).toBe('session-1'); + expect(JSON.parse(secondSessionKey)[0]).toBe('session-2'); + }); +}); diff --git a/ui/src/utils/monitorTopology.ts b/ui/src/utils/monitorTopology.ts new file mode 100644 index 000000000..fd60d4c26 --- /dev/null +++ b/ui/src/utils/monitorTopology.ts @@ -0,0 +1,8 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +export const buildMonitorTopologyKey = ( + selectedSessionId: string | null, + topologyFingerprint: string +): string => JSON.stringify([selectedSessionId, topologyFingerprint]); diff --git a/ui/src/views/MonitorView.tsx b/ui/src/views/MonitorView.tsx index d2f3b3ffa..18ab55558 100644 --- a/ui/src/views/MonitorView.tsx +++ b/ui/src/views/MonitorView.tsx @@ -87,6 +87,7 @@ import { import { deepMergeSchemas, validateValue } from '@/utils/jsonSchema'; import type { JsonSchema, JsonSchemaProperty } from '@/utils/jsonSchema'; import { viewsLogger } from '@/utils/logger'; +import { buildMonitorTopologyKey } from '@/utils/monitorTopology'; import { buildEdgesFromConnections, buildNodeObject, @@ -467,10 +468,11 @@ const MonitorViewContent: React.FC = () => { const draftFingerprint = Array.from(draftNodes.entries()) .map(([id, d]) => `${id}:${d.kind}:${d.missingRequired.join(',')}:${d.inFlight ? '1' : '0'}`) .sort(); - const key = JSON.stringify([kinds, conns, runtimeKeys, draftFingerprint]); + const topologyFingerprint = JSON.stringify([kinds, conns, runtimeKeys, draftFingerprint]); + const key = buildMonitorTopologyKey(selectedSessionId, topologyFingerprint); viewsLogger.debug('topoKey recalculated:', key.substring(0, 100)); return key; - }, [pipeline, draftNodes]); + }, [pipeline, draftNodes, selectedSessionId]); const { setNeedsAutoLayout, setNeedsFit, handleAutoLayout } = useAutoLayout({ pipeline, From 211aa47f5deeee0fc6bf324b1c9cae6e707bce87 Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 16:41:10 +0000 Subject: [PATCH 4/7] fix(ui): order monitor imports Signed-off-by: streamkit-devin --- ui/src/components/TypedEdge.tsx | 4 ++-- ui/src/components/monitor.render-perf.test.tsx | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/ui/src/components/TypedEdge.tsx b/ui/src/components/TypedEdge.tsx index 4380dc4ac..b8c496cf6 100644 --- a/ui/src/components/TypedEdge.tsx +++ b/ui/src/components/TypedEdge.tsx @@ -17,14 +17,14 @@ import React from 'react'; import { SKTooltip } from '@/components/Tooltip'; import { nodeKey, nodeStateAtom } from '@/stores/sessionAtoms'; import type { PacketType } from '@/types/types'; +import { deepEqual } from '@/utils/deepEqual'; +import { getPacketTypeColor } from '@/utils/packetTypes'; import { describeSlowInputsFromConnections, extractSlowTimeoutDetailsFromNodeState, type MonitorEdgeAlertContext, type SlowTimeoutDetails, } from '@/utils/pipelineGraph'; -import { deepEqual } from '@/utils/deepEqual'; -import { getPacketTypeColor } from '@/utils/packetTypes'; export type TypedEdgeData = { resolvedType?: PacketType; diff --git a/ui/src/components/monitor.render-perf.test.tsx b/ui/src/components/monitor.render-perf.test.tsx index c602dc654..70be38b7c 100644 --- a/ui/src/components/monitor.render-perf.test.tsx +++ b/ui/src/components/monitor.render-perf.test.tsx @@ -7,8 +7,8 @@ import { ReactFlowProvider, useStoreApi, type Node } from '@xyflow/react'; import React, { useEffect } from 'react'; import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { PinRow } from '@/components/node/PinRow'; import { PinHandle } from '@/components/node/PinHandle'; +import { PinRow } from '@/components/node/PinRow'; import { sessionStore, nodeKey, nodeStateAtom } from '@/stores/sessionAtoms'; import type { Connection, NodeState, OutputPin } from '@/types/types'; From 1d4f9d93c95cfb8fbd78d02f076586e3934d921a Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 16:57:52 +0000 Subject: [PATCH 5/7] fix(ui): satisfy react doctor hook detection Signed-off-by: streamkit-devin --- ui/src/components/TypedEdge.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ui/src/components/TypedEdge.tsx b/ui/src/components/TypedEdge.tsx index b8c496cf6..98920c0bd 100644 --- a/ui/src/components/TypedEdge.tsx +++ b/ui/src/components/TypedEdge.tsx @@ -12,7 +12,7 @@ import { import { atom, type Atom } from 'jotai'; import { useAtomValue } from 'jotai/react'; import { selectAtom } from 'jotai/utils'; -import React from 'react'; +import React, { useMemo } from 'react'; import { SKTooltip } from '@/components/Tooltip'; import { nodeKey, nodeStateAtom } from '@/stores/sessionAtoms'; @@ -92,7 +92,7 @@ export function useSlowInputAlert( ): TypedEdgeAlert | null { const sessionId = monitorAlertContext?.sessionId; const targetHandle = edge.targetHandle ?? ''; - const detailsAtom = React.useMemo(() => { + const detailsAtom = useMemo(() => { if (!sessionId) return nullSlowInputDetailsAtom; return selectAtom( nodeStateAtom(nodeKey(sessionId, edge.target)), From ec765c7f07bf9512fed7f9655cc34348cf773078 Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 17:05:32 +0000 Subject: [PATCH 6/7] fix(ui): restore monitor positions per session Signed-off-by: streamkit-devin --- ui/src/utils/monitorTopology.test.ts | 26 ++++++++++++++++++++++++- ui/src/utils/monitorTopology.ts | 13 +++++++++++++ ui/src/views/MonitorView.tsx | 29 +++++++++++++--------------- 3 files changed, 51 insertions(+), 17 deletions(-) diff --git a/ui/src/utils/monitorTopology.test.ts b/ui/src/utils/monitorTopology.test.ts index cd193619e..e6cc90fd2 100644 --- a/ui/src/utils/monitorTopology.test.ts +++ b/ui/src/utils/monitorTopology.test.ts @@ -4,7 +4,7 @@ import { describe, expect, it } from 'vitest'; -import { buildMonitorTopologyKey } from './monitorTopology'; +import { buildMonitorTopologyKey, resolveMonitorNodePosition } from './monitorTopology'; describe('buildMonitorTopologyKey', () => { it('distinguishes identical topology fingerprints across sessions', () => { @@ -22,4 +22,28 @@ describe('buildMonitorTopologyKey', () => { expect(JSON.parse(firstSessionKey)[0]).toBe('session-1'); expect(JSON.parse(secondSessionKey)[0]).toBe('session-2'); }); + + it('preserves live positions for same-session topology rebuilds', () => { + const previousPositions = new Map([['node', { x: 12, y: 24 }]]); + const savedPositions = { node: { x: 48, y: 96 } }; + + expect(resolveMonitorNodePosition('node', true, previousPositions, savedPositions)).toEqual({ + x: 12, + y: 24, + }); + }); + + it("uses the target session's saved position after a cross-session rebuild", () => { + const previousPositions = new Map([['node', { x: 12, y: 24 }]]); + const savedPositions = { node: { x: 48, y: 96 } }; + + expect(resolveMonitorNodePosition('node', false, previousPositions, savedPositions)).toEqual({ + x: 48, + y: 96, + }); + }); + + it('defaults to the origin when no position is available', () => { + expect(resolveMonitorNodePosition('node', false, new Map(), {})).toEqual({ x: 0, y: 0 }); + }); }); diff --git a/ui/src/utils/monitorTopology.ts b/ui/src/utils/monitorTopology.ts index fd60d4c26..889f41f39 100644 --- a/ui/src/utils/monitorTopology.ts +++ b/ui/src/utils/monitorTopology.ts @@ -6,3 +6,16 @@ export const buildMonitorTopologyKey = ( selectedSessionId: string | null, topologyFingerprint: string ): string => JSON.stringify([selectedSessionId, topologyFingerprint]); + +export type MonitorNodePosition = { x: number; y: number }; + +export const resolveMonitorNodePosition = ( + nodeName: string, + reusePreviousPositions: boolean, + previousPositions: ReadonlyMap, + savedPositions: Readonly> +): MonitorNodePosition => + (reusePreviousPositions ? previousPositions.get(nodeName) : savedPositions[nodeName]) ?? { + x: 0, + y: 0, + }; diff --git a/ui/src/views/MonitorView.tsx b/ui/src/views/MonitorView.tsx index 18ab55558..63a3cfc93 100644 --- a/ui/src/views/MonitorView.tsx +++ b/ui/src/views/MonitorView.tsx @@ -87,7 +87,7 @@ import { import { deepMergeSchemas, validateValue } from '@/utils/jsonSchema'; import type { JsonSchema, JsonSchemaProperty } from '@/utils/jsonSchema'; import { viewsLogger } from '@/utils/logger'; -import { buildMonitorTopologyKey } from '@/utils/monitorTopology'; +import { buildMonitorTopologyKey, resolveMonitorNodePosition } from '@/utils/monitorTopology'; import { buildEdgesFromConnections, buildNodeObject, @@ -756,18 +756,7 @@ const MonitorViewContent: React.FC = () => { ); const prevTopoKeyForTopologyRef = useRef(''); - - const resolveNodePosition = useCallback( - ( - nodeName: string, - prevPositions: Map, - savedPositions: Record - ): { position: { x: number; y: number } } => { - const pos = prevPositions.get(nodeName) ?? savedPositions[nodeName]; - return { position: pos ?? { x: 0, y: 0 } }; - }, - [] - ); + const topologySessionIdRef = useRef(null); const reconstructDynamicInputs = useCallback( ( @@ -946,6 +935,8 @@ const MonitorViewContent: React.FC = () => { return; } prevTopoKeyForTopologyRef.current = topoKey; + const reusePreviousPositions = topologySessionIdRef.current === selectedSessionId; + topologySessionIdRef.current = selectedSessionId; if (!pipeline && draftNodes.size === 0) { viewsLogger.debug('Topology effect: No pipeline, clearing nodes'); @@ -976,7 +967,12 @@ const MonitorViewContent: React.FC = () => { const apiNode = pipeline!.nodes[nodeName]; if (!apiNode) continue; - const { position: pos } = resolveNodePosition(nodeName, prevPositions, savedPositions); + const pos = resolveMonitorNodePosition( + nodeName, + reusePreviousPositions, + prevPositions, + savedPositions + ); const nodeState = (selectedSessionId @@ -1029,7 +1025,9 @@ const MonitorViewContent: React.FC = () => { const draftBaseOutputs = draftDef?.outputs ?? []; const draftFinalInputs = draftBaseInputs; const draftFinalOutputs = draftBaseOutputs; - const draftPos = prevPositions.get(draftId) ?? savedPositions[draftId] ?? draft.position; + const draftPos = + (reusePreviousPositions ? prevPositions.get(draftId) : savedPositions[draftId]) ?? + draft.position; const node = buildNodeObject({ nodeName: draftId, apiNode: { @@ -1089,7 +1087,6 @@ const MonitorViewContent: React.FC = () => { selectedSessionId, getNodePositions, defByKind, - resolveNodePosition, resolveDynamicPins, stableOnParamChange, stableOnConfigChange, From 7da53d77d9538916285f4ea9c3c857464e1f4a19 Mon Sep 17 00:00:00 2001 From: streamkit-devin Date: Sun, 23 Aug 2026 17:06:22 +0000 Subject: [PATCH 7/7] fix(ui): preserve saved monitor positions Signed-off-by: streamkit-devin --- ui/src/utils/monitorTopology.test.ts | 10 ++++++++++ ui/src/utils/monitorTopology.ts | 6 ++---- ui/src/views/MonitorView.tsx | 3 ++- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/ui/src/utils/monitorTopology.test.ts b/ui/src/utils/monitorTopology.test.ts index e6cc90fd2..de67861b6 100644 --- a/ui/src/utils/monitorTopology.test.ts +++ b/ui/src/utils/monitorTopology.test.ts @@ -33,6 +33,16 @@ describe('buildMonitorTopologyKey', () => { }); }); + it('uses the saved position for same-session nodes without a live position', () => { + const previousPositions = new Map(); + const savedPositions = { node: { x: 48, y: 96 } }; + + expect(resolveMonitorNodePosition('node', true, previousPositions, savedPositions)).toEqual({ + x: 48, + y: 96, + }); + }); + it("uses the target session's saved position after a cross-session rebuild", () => { const previousPositions = new Map([['node', { x: 12, y: 24 }]]); const savedPositions = { node: { x: 48, y: 96 } }; diff --git a/ui/src/utils/monitorTopology.ts b/ui/src/utils/monitorTopology.ts index 889f41f39..476da6fb9 100644 --- a/ui/src/utils/monitorTopology.ts +++ b/ui/src/utils/monitorTopology.ts @@ -15,7 +15,5 @@ export const resolveMonitorNodePosition = ( previousPositions: ReadonlyMap, savedPositions: Readonly> ): MonitorNodePosition => - (reusePreviousPositions ? previousPositions.get(nodeName) : savedPositions[nodeName]) ?? { - x: 0, - y: 0, - }; + (reusePreviousPositions ? previousPositions.get(nodeName) : undefined) ?? + savedPositions[nodeName] ?? { x: 0, y: 0 }; diff --git a/ui/src/views/MonitorView.tsx b/ui/src/views/MonitorView.tsx index 63a3cfc93..de51fe2c3 100644 --- a/ui/src/views/MonitorView.tsx +++ b/ui/src/views/MonitorView.tsx @@ -1026,7 +1026,8 @@ const MonitorViewContent: React.FC = () => { const draftFinalInputs = draftBaseInputs; const draftFinalOutputs = draftBaseOutputs; const draftPos = - (reusePreviousPositions ? prevPositions.get(draftId) : savedPositions[draftId]) ?? + (reusePreviousPositions ? prevPositions.get(draftId) : undefined) ?? + savedPositions[draftId] ?? draft.position; const node = buildNodeObject({ nodeName: draftId,