|
|
@@ -48,6 +48,7 @@ import { WorkflowNode, NodeExecutionState, IterationResult, resolveOutputs } fro
|
|
|
import { ExecutionResultsPanel } from '../components/workflow/ExecutionResultsPanel'
|
|
|
import { NodePickerSidebar } from '../components/workflow/NodePickerSidebar'
|
|
|
import { DeleteConfirmModal } from '../components/workflow/DeleteConfirmModal'
|
|
|
+import { ConfirmModal } from '../components/workflow/ConfirmModal'
|
|
|
import { NodeContextMenu, EdgeContextMenu } from '../components/workflow/ContextMenus'
|
|
|
import { ToastContainer, ToastMessage } from '../components/workflow/Toast'
|
|
|
import { EditorHeader } from '../components/workflow/EditorHeader'
|
|
|
@@ -55,6 +56,7 @@ import { WorkflowSettingsModal } from '../components/workflow/WorkflowSettingsMo
|
|
|
import { NodeConfigModal } from '../components/workflow/NodeConfigModal'
|
|
|
import { copyNodesToClipboard, readNodeClipboard, offsetFor } from '../utils/nodeClipboard'
|
|
|
import { takenNodeNames, uniqueNodeName } from '../utils/nodeNames'
|
|
|
+import { computeNodeToLoopMap } from '../utils/loopMembership'
|
|
|
import { useGraphHistory } from '../hooks/useGraphHistory'
|
|
|
|
|
|
interface SelectedNodeData {
|
|
|
@@ -80,6 +82,27 @@ interface ExecutionState {
|
|
|
// back from the stored record for a pinned/viewed one; absent for a normal
|
|
|
// run and for anything recorded before the engine sent this field.
|
|
|
singleNodeTarget?: string
|
|
|
+ // The loop iteration that single-node run targeted, when the node ran
|
|
|
+ // inside a loop. See ExecutionDetail.singleNodeIteration - absent means
|
|
|
+ // iteration 0 or an engine that predates the field, and the panel treats
|
|
|
+ // both the same way.
|
|
|
+ singleNodeIteration?: number
|
|
|
+}
|
|
|
+
|
|
|
+// A re-run launched from a row in the results panel, tracked entirely apart
|
|
|
+// from executionState/pinnedExecution so it can never overwrite either.
|
|
|
+// Discarded on navigation (it lives only in this component's state) and
|
|
|
+// never written to the stored execution - see the design doc's "where does
|
|
|
+// the result go" section.
|
|
|
+interface RerunState {
|
|
|
+ nodeId: string
|
|
|
+ iteration?: number
|
|
|
+ status: 'running' | 'completed' | 'failed'
|
|
|
+ output?: any
|
|
|
+ error?: string
|
|
|
+ startedAt: number
|
|
|
+ completedAt?: number
|
|
|
+ executionId?: string
|
|
|
}
|
|
|
|
|
|
interface StorageCollection {
|
|
|
@@ -417,6 +440,16 @@ function WorkflowEditorInner() {
|
|
|
// Selected iteration index per node (for viewing specific iteration results in loop body nodes)
|
|
|
const [selectedIterations, setSelectedIterations] = useState<Record<string, number>>({})
|
|
|
|
|
|
+ // A re-run launched from the results panel: in flight or just finished,
|
|
|
+ // never more than one at a time (starting another replaces it). Entirely
|
|
|
+ // separate from executionState/pinnedExecution - see RerunState above.
|
|
|
+ const [rerunState, setRerunState] = useState<RerunState | null>(null)
|
|
|
+ const pendingRerunExecIdRef = useRef<string | null>(null)
|
|
|
+ // A re-run the user has asked for but not yet confirmed - the node names
|
|
|
+ // itself and the fact that this executes it for real (side effects
|
|
|
+ // included) before anything actually runs.
|
|
|
+ const [rerunConfirm, setRerunConfirm] = useState<{ nodeId: string; iteration?: number } | null>(null)
|
|
|
+
|
|
|
// Workflow settings state
|
|
|
const [showWorkflowSettings, setShowWorkflowSettings] = useState(false)
|
|
|
const [workflowSettings, setWorkflowSettings] = useState<Record<string, any>>({})
|
|
|
@@ -437,6 +470,14 @@ function WorkflowEditorInner() {
|
|
|
recomputeChangeStatus,
|
|
|
} = useExecutionPinStore()
|
|
|
|
|
|
+ // Discard the scratch re-run whenever what's being looked at changes -
|
|
|
+ // a different workflow, a different pinned execution, or leaving the
|
|
|
+ // pinned view entirely. "Discarded on navigation" per the design.
|
|
|
+ useEffect(() => {
|
|
|
+ setRerunState(null)
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ }, [id, pinnedExecution?.id, isViewingExecution])
|
|
|
+
|
|
|
// A record's status as the engine wrote it, narrowed to what the panel and
|
|
|
// canvas actually distinguish. Unrecognized values fall back to
|
|
|
// 'completed' rather than disappearing, since that was the pre-existing
|
|
|
@@ -551,6 +592,7 @@ function WorkflowEditorInner() {
|
|
|
// handlers below and never populates this.
|
|
|
rawNodeExecutions: sortedExecutions,
|
|
|
singleNodeTarget: pinnedExecution.singleNodeTarget,
|
|
|
+ singleNodeIteration: pinnedExecution.singleNodeIteration,
|
|
|
}
|
|
|
}
|
|
|
return executionState
|
|
|
@@ -1072,48 +1114,99 @@ function WorkflowEditorInner() {
|
|
|
return JSON.stringify(config || {})
|
|
|
}, [])
|
|
|
|
|
|
- // Execute workflow up to a specific node (run single node)
|
|
|
- const executeNode = useCallback((targetNodeId: string) => {
|
|
|
- if (!id) return
|
|
|
-
|
|
|
- // Build cached outputs for upstream nodes that have unchanged configs
|
|
|
- // This allows the backend to skip re-executing nodes that haven't changed
|
|
|
- const cachedOutputs: Record<string, { output: any; configHash: string }> = {}
|
|
|
-
|
|
|
- // Collect all upstream node IDs (nodes that feed into the target)
|
|
|
- const collectUpstreamNodes = (nodeId: string, visited: Set<string> = new Set()): string[] => {
|
|
|
+ // All node IDs upstream of targetNodeId, walking the live draft graph -
|
|
|
+ // the same graph a re-run's target node config comes from, so "upstream"
|
|
|
+ // means the same thing as "what the target is currently wired to".
|
|
|
+ const collectUpstreamNodes = useCallback((targetNodeId: string): string[] => {
|
|
|
+ const visited = new Set<string>()
|
|
|
+ const walk = (nodeId: string): string[] => {
|
|
|
if (visited.has(nodeId)) return []
|
|
|
visited.add(nodeId)
|
|
|
-
|
|
|
const upstreamIds: string[] = []
|
|
|
for (const edge of edges) {
|
|
|
if (edge.target === nodeId) {
|
|
|
upstreamIds.push(edge.source)
|
|
|
- upstreamIds.push(...collectUpstreamNodes(edge.source, visited))
|
|
|
+ upstreamIds.push(...walk(edge.source))
|
|
|
}
|
|
|
}
|
|
|
return upstreamIds
|
|
|
}
|
|
|
+ return walk(targetNodeId)
|
|
|
+ }, [edges])
|
|
|
|
|
|
+ // Build the cachedOutputs map sent alongside a single-node run.
|
|
|
+ //
|
|
|
+ // fromExecution given: scope strictly to that one execution's own
|
|
|
+ // nodeExecutions records - matching loop iteration when the upstream node
|
|
|
+ // shares the target's loop context - so the re-run reproduces exactly
|
|
|
+ // what that execution saw, not a blend of several runs. This is what "a
|
|
|
+ // re-run launched from a specific execution's detail view" means.
|
|
|
+ //
|
|
|
+ // fromExecution omitted: the original heuristic, unchanged - whatever is
|
|
|
+ // cached in this editor session (lastExecutionResults, seeded from the
|
|
|
+ // last five executions) or pinned by hand on the canvas. Stays the
|
|
|
+ // default for "just try the current draft", where there is no one
|
|
|
+ // execution to be faithful to.
|
|
|
+ const buildCachedOutputs = useCallback((
|
|
|
+ targetNodeId: string,
|
|
|
+ fromExecution?: ExecutionDetail,
|
|
|
+ iteration?: number
|
|
|
+ ): Record<string, { output: any; configHash: string }> => {
|
|
|
+ const cachedOutputs: Record<string, { output: any; configHash: string }> = {}
|
|
|
const upstreamNodeIds = collectUpstreamNodes(targetNodeId)
|
|
|
|
|
|
- // For each upstream node, if we have cached output, include it. Data pinned
|
|
|
- // from an earlier execution counts as cached: pinning it is how you say
|
|
|
- // "use this instead of running the workflow again", and without sending it
|
|
|
- // every upstream node would run for real.
|
|
|
+ if (fromExecution) {
|
|
|
+ const nodeToLoopMap = computeNodeToLoopMap(nodes, edges)
|
|
|
+ const targetLoopId = nodeToLoopMap.get(targetNodeId)
|
|
|
+ const entries = fromExecution.nodeExecutions || []
|
|
|
+ for (const nodeId of upstreamNodeIds) {
|
|
|
+ const upstreamLoopId = nodeToLoopMap.get(nodeId)
|
|
|
+ const sameLoopContext = upstreamLoopId !== undefined && upstreamLoopId === targetLoopId
|
|
|
+ const candidates = entries.filter(e => e.nodeId === nodeId && e.status === 'completed' && e.output !== undefined)
|
|
|
+
|
|
|
+ // Prefer the exact iteration the target ran in, when this upstream
|
|
|
+ // node was part of the same loop body. Otherwise a plain, non-loop
|
|
|
+ // record. Otherwise - an upstream node from a different loop
|
|
|
+ // context, with no "selected iteration" to match - the most
|
|
|
+ // recently completed record for it, which is the least surprising
|
|
|
+ // fallback absent an iteration to be faithful to.
|
|
|
+ let chosen = sameLoopContext
|
|
|
+ ? candidates.filter(e => e.loopNodeId === upstreamLoopId && e.loopIteration === iteration).slice(-1)[0]
|
|
|
+ : undefined
|
|
|
+ if (!chosen) chosen = candidates.filter(e => e.loopNodeId === undefined).slice(-1)[0]
|
|
|
+ if (!chosen) chosen = candidates.slice(-1)[0]
|
|
|
+
|
|
|
+ if (chosen) {
|
|
|
+ const node = nodes.find(n => n.id === nodeId)
|
|
|
+ cachedOutputs[nodeId] = { output: chosen.output, configHash: hashConfig(node?.data?.config) }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return cachedOutputs
|
|
|
+ }
|
|
|
+
|
|
|
+ // Data pinned from an earlier execution counts as cached too: pinning it
|
|
|
+ // is how you say "use this instead of running the workflow again", and
|
|
|
+ // without sending it every upstream node would run for real.
|
|
|
for (const nodeId of upstreamNodeIds) {
|
|
|
const cachedOutput = lastExecutionResults[nodeId] ?? pinnedNodeData[nodeId]
|
|
|
if (cachedOutput !== undefined) {
|
|
|
- // Find the node to get its current config
|
|
|
const node = nodes.find(n => n.id === nodeId)
|
|
|
if (node) {
|
|
|
- cachedOutputs[nodeId] = {
|
|
|
- output: cachedOutput,
|
|
|
- configHash: hashConfig(node.data?.config),
|
|
|
- }
|
|
|
+ cachedOutputs[nodeId] = { output: cachedOutput, configHash: hashConfig(node.data?.config) }
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ return cachedOutputs
|
|
|
+ }, [collectUpstreamNodes, nodes, edges, lastExecutionResults, pinnedNodeData, hashConfig])
|
|
|
+
|
|
|
+ // Execute workflow up to a specific node (run single node). Unchanged
|
|
|
+ // behaviour from before this feature: the canvas's "Run This Node" action
|
|
|
+ // - the cache heuristic is the recent/pinned one (buildCachedOutputs with
|
|
|
+ // no fromExecution), never scoped to a specific execution.
|
|
|
+ const executeNode = useCallback((targetNodeId: string) => {
|
|
|
+ if (!id) return
|
|
|
+
|
|
|
+ const cachedOutputs = buildCachedOutputs(targetNodeId)
|
|
|
|
|
|
// Set pending marker BEFORE HTTP request to capture early WebSocket events
|
|
|
pendingExecIdRef.current = 'pending'
|
|
|
@@ -1151,7 +1244,58 @@ function WorkflowEditorInner() {
|
|
|
setExecutingNodeId(null)
|
|
|
showToast('error', `Execution failed: ${error.message}`)
|
|
|
})
|
|
|
- }, [id, showToast, nodes, edges, lastExecutionResults, pinnedNodeData, hashConfig])
|
|
|
+ }, [id, showToast, buildCachedOutputs])
|
|
|
+
|
|
|
+ // A re-run launched from a row in the results panel. Kept apart from
|
|
|
+ // executeNode/executionState on purpose - see RerunState above - so it
|
|
|
+ // can be shown next to whatever the panel already has on screen instead
|
|
|
+ // of replacing it.
|
|
|
+ const rerunNodeFromPanel = useCallback((targetNodeId: string, iteration?: number) => {
|
|
|
+ if (!id) return
|
|
|
+
|
|
|
+ // "The execution being inspected" - present whenever the panel is
|
|
|
+ // currently showing one specific past execution rather than a live
|
|
|
+ // run's own results. Mirrors effectiveExecutionState's own condition
|
|
|
+ // for using pinnedExecution, so the two never disagree about which
|
|
|
+ // execution is on screen.
|
|
|
+ const fromExecution = (pinnedExecution && (isViewingExecution || executionState.status === 'idle'))
|
|
|
+ ? pinnedExecution
|
|
|
+ : undefined
|
|
|
+
|
|
|
+ const cachedOutputs = buildCachedOutputs(targetNodeId, fromExecution, iteration)
|
|
|
+
|
|
|
+ pendingRerunExecIdRef.current = 'pending'
|
|
|
+ setRerunState({
|
|
|
+ nodeId: targetNodeId,
|
|
|
+ iteration,
|
|
|
+ status: 'running',
|
|
|
+ startedAt: Date.now(),
|
|
|
+ })
|
|
|
+
|
|
|
+ workflowsApi.executeToNode(id, targetNodeId, undefined, cachedOutputs, iteration).then((result) => {
|
|
|
+ pendingRerunExecIdRef.current = result.executionId
|
|
|
+ setRerunState((prev) => (prev && prev.nodeId === targetNodeId ? { ...prev, executionId: result.executionId } : prev))
|
|
|
+ }).catch((error) => {
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ setRerunState((prev) => (prev && prev.nodeId === targetNodeId
|
|
|
+ ? { ...prev, status: 'failed', error: error.message, completedAt: Date.now() }
|
|
|
+ : prev))
|
|
|
+ showToast('error', `Re-run failed: ${error.message}`)
|
|
|
+ })
|
|
|
+ }, [id, pinnedExecution, isViewingExecution, executionState.status, buildCachedOutputs, showToast])
|
|
|
+
|
|
|
+ // Ask before rerunNodeFromPanel actually runs anything - it executes the
|
|
|
+ // node for real, side effects included, and there is no reliable list of
|
|
|
+ // which nodes are safe (see nodeDefsMap-based message built near the
|
|
|
+ // ConfirmModal render below).
|
|
|
+ const requestRerun = useCallback((nodeId: string, iteration?: number) => {
|
|
|
+ setRerunConfirm({ nodeId, iteration })
|
|
|
+ }, [])
|
|
|
+
|
|
|
+ const dismissRerun = useCallback(() => {
|
|
|
+ setRerunState(null)
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ }, [])
|
|
|
|
|
|
// WebSocket subscription for execution updates - subscribe on mount, filter in handler
|
|
|
useEffect(() => {
|
|
|
@@ -1444,6 +1588,60 @@ function WorkflowEditorInner() {
|
|
|
}
|
|
|
}, [showToast]) // Subscribe on mount, unsubscribe on unmount
|
|
|
|
|
|
+ // A second, independent listener on the same channel for a panel-launched
|
|
|
+ // re-run. Kept apart from handleExecutionEvent above so a rerun in flight
|
|
|
+ // can never be confused with (or overwrite) a live full run's state - it
|
|
|
+ // updates only rerunState, and only for the node it targeted.
|
|
|
+ useEffect(() => {
|
|
|
+ const handleRerunEvent = (data: any) => {
|
|
|
+ const channel = data._channel || ''
|
|
|
+ const parts = channel.split('.')
|
|
|
+ if (parts.length < 3) return
|
|
|
+
|
|
|
+ const eventExecId = parts[1]
|
|
|
+ const eventType = parts.slice(2).join('.')
|
|
|
+
|
|
|
+ const targetExecId = pendingRerunExecIdRef.current
|
|
|
+ if (!targetExecId) return
|
|
|
+
|
|
|
+ if (targetExecId === 'pending') {
|
|
|
+ pendingRerunExecIdRef.current = eventExecId
|
|
|
+ } else if (eventExecId !== targetExecId) {
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ setRerunState((prev) => {
|
|
|
+ if (!prev) return prev
|
|
|
+ const isTarget = data.nodeId === prev.nodeId
|
|
|
+ if (isTarget && (eventType === 'node_completed' || eventType === 'loop_node_completed' || eventType === 'loop.node.completed')) {
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ return { ...prev, status: 'completed', output: data.output, completedAt: Date.now(), executionId: eventExecId }
|
|
|
+ }
|
|
|
+ if (isTarget && (eventType === 'node_failed' || eventType === 'loop_node_failed' || eventType === 'loop.node.failed')) {
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ return { ...prev, status: 'failed', error: data.error, completedAt: Date.now(), executionId: eventExecId }
|
|
|
+ }
|
|
|
+ // The whole targeted run failed before reaching the target node - an
|
|
|
+ // upstream node whose config changed (so it wasn't served from
|
|
|
+ // cache) failed for real, for instance.
|
|
|
+ if (eventType === 'failed' && prev.status === 'running') {
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ return { ...prev, status: 'failed', error: data.error || 'Execution failed before this node ran', completedAt: Date.now(), executionId: eventExecId }
|
|
|
+ }
|
|
|
+ if (eventType === 'completed' && prev.status === 'running') {
|
|
|
+ pendingRerunExecIdRef.current = null
|
|
|
+ return { ...prev, completedAt: Date.now(), executionId: eventExecId }
|
|
|
+ }
|
|
|
+ return prev
|
|
|
+ })
|
|
|
+ }
|
|
|
+
|
|
|
+ wsClient.on('executions.*', handleRerunEvent)
|
|
|
+ return () => {
|
|
|
+ wsClient.off('executions.*', handleRerunEvent)
|
|
|
+ }
|
|
|
+ }, [])
|
|
|
+
|
|
|
const nodeDefs: NodeDefinition[] = nodeDefinitions?.nodes || []
|
|
|
const nodeDefsMap = useMemo(() => {
|
|
|
const map: Record<string, NodeDefinition> = {}
|
|
|
@@ -1451,6 +1649,24 @@ function WorkflowEditorInner() {
|
|
|
return map
|
|
|
}, [nodeDefs])
|
|
|
|
|
|
+ // Message for the confirmation shown before rerunNodeFromPanel actually
|
|
|
+ // runs anything. There is no hardcoded list of "dangerous" node types
|
|
|
+ // here on purpose - that rots the moment a new node is added - so this
|
|
|
+ // stays honest instead: specific when the node declares sideEffects
|
|
|
+ // (not sent by the backend yet, so this branch is currently unreachable
|
|
|
+ // but ready for the day it is), generic and equally true otherwise.
|
|
|
+ const buildRerunConfirmMessage = useCallback((nodeId: string, iteration?: number): string => {
|
|
|
+ const node = nodes.find(n => n.id === nodeId)
|
|
|
+ const label = node?.data.config?._customLabel || node?.data.label || nodeId
|
|
|
+ const def = nodeDefsMap[node?.data?.type || '']
|
|
|
+ const iterText = iteration !== undefined ? ` as it ran in iteration ${iteration + 1}` : ''
|
|
|
+ const reuseNote = 'Upstream nodes are reused from the execution you\'re looking at where their config is unchanged; only this node (and anything whose config changed) actually runs.'
|
|
|
+ if (def?.sideEffects) {
|
|
|
+ return `"${label}" is marked as having side effects. Re-running it${iterText} executes it for real, using its current draft config - any external call, write, or publish it makes will happen again.\n\n${reuseNote}`
|
|
|
+ }
|
|
|
+ return `Re-running "${label}"${iterText} executes it for real, using its current draft config. If it calls an external service, writes data, or has any other side effect, that happens again - this cannot tell whether it does.\n\n${reuseNote}`
|
|
|
+ }, [nodes, nodeDefsMap])
|
|
|
+
|
|
|
// What every editor watching this workflow should currently have. Edits are
|
|
|
// sent as the difference from this, and it moves forward both when this
|
|
|
// canvas broadcasts and when somebody else's edit is applied.
|
|
|
@@ -3152,6 +3368,7 @@ function WorkflowEditorInner() {
|
|
|
expandedNodes={expandedNodes}
|
|
|
selectedIterations={selectedIterations}
|
|
|
isViewingExecution={isViewingExecution}
|
|
|
+ rerunResult={rerunState}
|
|
|
onClose={() => {
|
|
|
// Closing here means exactly the same thing as the banner's
|
|
|
// "Return to Editor" - both go through handleReturnToEditor, so
|
|
|
@@ -3167,10 +3384,30 @@ function WorkflowEditorInner() {
|
|
|
onSelectIteration={(nodeId, index) => {
|
|
|
setSelectedIterations(prev => ({...prev, [nodeId]: index}))
|
|
|
}}
|
|
|
+ onRerunNode={requestRerun}
|
|
|
+ onDismissRerun={dismissRerun}
|
|
|
/>
|
|
|
)}
|
|
|
</div>
|
|
|
|
|
|
+ {/* Re-run confirmation - the node executes for real, side effects
|
|
|
+ included, so this is asked every time rather than only for nodes
|
|
|
+ this page happens to guess are risky. */}
|
|
|
+ {rerunConfirm && (
|
|
|
+ <ConfirmModal
|
|
|
+ title="Re-run this node?"
|
|
|
+ message={buildRerunConfirmMessage(rerunConfirm.nodeId, rerunConfirm.iteration)}
|
|
|
+ confirmLabel="Re-run"
|
|
|
+ tone="warning"
|
|
|
+ onConfirm={() => {
|
|
|
+ const { nodeId, iteration } = rerunConfirm
|
|
|
+ setRerunConfirm(null)
|
|
|
+ rerunNodeFromPanel(nodeId, iteration)
|
|
|
+ }}
|
|
|
+ onCancel={() => setRerunConfirm(null)}
|
|
|
+ />
|
|
|
+ )}
|
|
|
+
|
|
|
{/* Delete confirmation modal */}
|
|
|
{showDeleteConfirm && (
|
|
|
<DeleteConfirmModal
|
