fix(copilot): fix execute workflow from diff store (#1894)

* Fix run from diff store

* Fix copilot run workflow
This commit is contained in:
Siddharth Ganesan
2025-11-11 11:09:17 -08:00
committed by GitHub
parent 41f3d506da
commit c86f2a0537
6 changed files with 222 additions and 14 deletions
@@ -30,6 +30,15 @@ const ExecuteWorkflowSchema = z.object({
useDraftState: z.boolean().optional(),
input: z.any().optional(),
startBlockId: z.string().optional(),
// Optional workflow state override (for executing diff workflows)
workflowStateOverride: z
.object({
blocks: z.record(z.any()),
edges: z.array(z.any()),
loops: z.record(z.any()).optional(),
parallels: z.record(z.any()).optional(),
})
.optional(),
})
export const runtime = 'nodejs'
@@ -310,6 +319,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
stream: streamParam,
useDraftState,
input: validatedInput,
workflowStateOverride,
} = validation.data
// For API key auth, the entire body is the input (except for our control fields)
@@ -317,7 +327,14 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
const input =
auth.authType === 'api_key'
? (() => {
const { selectedOutputs, triggerType, stream, useDraftState, ...rest } = body
const {
selectedOutputs,
triggerType,
stream,
useDraftState,
workflowStateOverride,
...rest
} = body
return Object.keys(rest).length > 0 ? rest : validatedInput
})()
: validatedInput
@@ -460,6 +477,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
triggerType,
useDraftState: shouldUseDraftState,
startTime: new Date().toISOString(),
workflowStateOverride,
}
const snapshot = new ExecutionSnapshot(
@@ -714,6 +732,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
triggerType,
useDraftState: shouldUseDraftState,
startTime: new Date().toISOString(),
workflowStateOverride,
}
const snapshot = new ExecutionSnapshot(
@@ -1,5 +1,6 @@
import { useCallback, useState } from 'react'
import { v4 as uuidv4 } from 'uuid'
import { shallow } from 'zustand/shallow'
import { createLogger } from '@/lib/logs/console/logger'
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
import { processStreamingBlockLogs } from '@/lib/tokenization'
@@ -16,6 +17,7 @@ import { useExecutionStore } from '@/stores/execution/store'
import { useVariablesStore } from '@/stores/panel/variables/store'
import { useEnvironmentStore } from '@/stores/settings/environment/store'
import { useTerminalConsoleStore } from '@/stores/terminal'
import { useWorkflowDiffStore } from '@/stores/workflow-diff'
import { useWorkflowRegistry } from '@/stores/workflows/registry/store'
import { mergeSubblockState } from '@/stores/workflows/utils'
import { useCurrentWorkflow } from './use-current-workflow'
@@ -99,6 +101,26 @@ export function useWorkflowExecution() {
} = useExecutionStore()
const [executionResult, setExecutionResult] = useState<ExecutionResult | null>(null)
const executionStream = useExecutionStream()
const {
diffWorkflow: executionDiffWorkflow,
isDiffReady: isDiffWorkflowReady,
isShowingDiff: isViewingDiff,
} = useWorkflowDiffStore(
useCallback(
(state) => ({
diffWorkflow: state.diffWorkflow,
isDiffReady: state.isDiffReady,
isShowingDiff: state.isShowingDiff,
}),
[]
),
shallow
)
const hasActiveDiffWorkflow =
isDiffWorkflowReady &&
isViewingDiff &&
!!executionDiffWorkflow &&
Object.keys(executionDiffWorkflow.blocks || {}).length > 0
/**
* Validates debug state before performing debug operations
@@ -645,8 +667,14 @@ export function useWorkflowExecution() {
onBlockComplete?: (blockId: string, output: any) => Promise<void>,
overrideTriggerType?: 'chat' | 'manual' | 'api'
): Promise<ExecutionResult | StreamingExecution> => {
// Use currentWorkflow but check if we're in diff mode
const { blocks: workflowBlocks, edges: workflowEdges } = currentWorkflow
// Use diff workflow for execution when available, regardless of canvas view state
const executionWorkflowState =
hasActiveDiffWorkflow && executionDiffWorkflow ? executionDiffWorkflow : null
const usingDiffForExecution = executionWorkflowState !== null
const workflowBlocks = (executionWorkflowState?.blocks ??
currentWorkflow.blocks) as typeof currentWorkflow.blocks
const workflowEdges = (executionWorkflowState?.edges ??
currentWorkflow.edges) as typeof currentWorkflow.edges
// Filter out blocks without type (these are layout-only blocks)
const validBlocks = Object.entries(workflowBlocks).reduce(
@@ -665,6 +693,9 @@ export function useWorkflowExecution() {
logger.info('Executing workflow', {
isDiffMode: currentWorkflow.isDiffMode,
usingDiffForExecution,
isViewingDiff,
executingDiffWorkflow: usingDiffForExecution && isViewingDiff,
isExecutingFromChat,
totalBlocksCount: Object.keys(workflowBlocks).length,
validBlocksCount: Object.keys(validBlocks).length,
@@ -838,6 +869,15 @@ export function useWorkflowExecution() {
selectedOutputs,
triggerType: overrideTriggerType || 'manual',
useDraftState: true,
// Pass diff workflow state if available for execution
workflowStateOverride: executionWorkflowState
? {
blocks: executionWorkflowState.blocks,
edges: executionWorkflowState.edges,
loops: executionWorkflowState.loops,
parallels: executionWorkflowState.parallels,
}
: undefined,
callbacks: {
onExecutionStarted: (data) => {
logger.info('Server execution started:', data)
@@ -1,4 +1,7 @@
import { v4 as uuidv4 } from 'uuid'
import type { ExecutionResult, StreamingExecution } from '@/executor/types'
import { useTerminalConsoleStore } from '@/stores/terminal'
import { useWorkflowDiffStore } from '@/stores/workflow-diff/store'
import { useWorkflowRegistry } from '@/stores/workflows/registry/store'
export interface WorkflowExecutionOptions {
@@ -11,7 +14,7 @@ export interface WorkflowExecutionOptions {
/**
* Execute workflow with full logging (used by copilot tools)
* This now delegates to the server-side executor via API
* Handles SSE streaming and populates console logs in real-time
*/
export async function executeWorkflowWithFullLogging(
options: WorkflowExecutionOptions = {}
@@ -22,18 +25,41 @@ export async function executeWorkflowWithFullLogging(
throw new Error('No active workflow')
}
// For copilot tool calls, we use non-SSE execution to get a simple result
// Check if there's an active diff workflow to execute
const { diffWorkflow, isDiffReady, isShowingDiff } = useWorkflowDiffStore.getState()
const hasActiveDiffWorkflow =
isDiffReady &&
isShowingDiff &&
!!diffWorkflow &&
Object.keys(diffWorkflow.blocks || {}).length > 0
const executionId = options.executionId || uuidv4()
const { addConsole } = useTerminalConsoleStore.getState()
// Build request payload
const payload: any = {
input: options.workflowInput,
stream: true,
triggerType: options.overrideTriggerType || 'manual',
useDraftState: true,
}
// Add diff workflow override if active
if (hasActiveDiffWorkflow) {
payload.workflowStateOverride = {
blocks: diffWorkflow.blocks,
edges: diffWorkflow.edges,
loops: diffWorkflow.loops,
parallels: diffWorkflow.parallels,
}
}
const response = await fetch(`/api/workflows/${activeWorkflowId}/execute`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
input: options.workflowInput,
stream: false, // Copilot doesn't need SSE streaming
triggerType: options.overrideTriggerType || 'manual',
useDraftState: true,
}),
body: JSON.stringify(payload),
})
if (!response.ok) {
@@ -41,6 +67,106 @@ export async function executeWorkflowWithFullLogging(
throw new Error(error.error || 'Workflow execution failed')
}
const result = await response.json()
return result as ExecutionResult
if (!response.body) {
throw new Error('No response body')
}
// Parse SSE stream
const reader = response.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
let executionResult: ExecutionResult = {
success: false,
output: {},
logs: [],
}
try {
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
const lines = buffer.split('\n\n')
buffer = lines.pop() || ''
for (const line of lines) {
if (!line.trim() || !line.startsWith('data: ')) continue
const data = line.substring(6).trim()
if (data === '[DONE]') continue
try {
const event = JSON.parse(data)
switch (event.type) {
case 'block:completed':
addConsole({
input: event.data.input || {},
output: event.data.output,
success: true,
durationMs: event.data.durationMs,
startedAt: new Date(Date.now() - event.data.durationMs).toISOString(),
endedAt: new Date().toISOString(),
workflowId: activeWorkflowId,
blockId: event.data.blockId,
executionId,
blockName: event.data.blockName,
blockType: event.data.blockType,
iterationCurrent: event.data.iterationCurrent,
iterationTotal: event.data.iterationTotal,
iterationType: event.data.iterationType,
})
if (options.onBlockComplete) {
options.onBlockComplete(event.data.blockId, event.data.output).catch(() => {})
}
break
case 'block:error':
addConsole({
input: event.data.input || {},
output: {},
success: false,
error: event.data.error,
durationMs: event.data.durationMs,
startedAt: new Date(Date.now() - event.data.durationMs).toISOString(),
endedAt: new Date().toISOString(),
workflowId: activeWorkflowId,
blockId: event.data.blockId,
executionId,
blockName: event.data.blockName,
blockType: event.data.blockType,
iterationCurrent: event.data.iterationCurrent,
iterationTotal: event.data.iterationTotal,
iterationType: event.data.iterationType,
})
break
case 'execution:completed':
executionResult = {
success: event.data.success,
output: event.data.output,
logs: [],
metadata: {
duration: event.data.duration,
startTime: event.data.startTime,
endTime: event.data.endTime,
},
}
break
case 'execution:error':
throw new Error(event.data.error || 'Execution failed')
}
} catch (parseError) {
// Skip malformed SSE events
}
}
}
} finally {
reader.releaseLock()
}
return executionResult
}
+6
View File
@@ -13,6 +13,12 @@ export interface ExecutionMetadata {
startTime: string
pendingBlocks?: string[]
resumeFromSnapshot?: boolean
workflowStateOverride?: {
blocks: Record<string, any>
edges: Edge[]
loops?: Record<string, any>
parallels?: Record<string, any>
}
}
export interface ExecutionCallbacks {
+6
View File
@@ -61,6 +61,12 @@ export interface ExecuteStreamOptions {
startBlockId?: string
triggerType?: string
useDraftState?: boolean
workflowStateOverride?: {
blocks: Record<string, any>
edges: any[]
loops?: Record<string, any>
parallels?: Record<string, any>
}
callbacks?: ExecutionStreamCallbacks
}
@@ -114,7 +114,18 @@ export async function executeWorkflowCore(
let loops
let parallels
if (useDraftState) {
// Use workflowStateOverride if provided (for diff workflows)
if (metadata.workflowStateOverride) {
blocks = metadata.workflowStateOverride.blocks
edges = metadata.workflowStateOverride.edges
loops = metadata.workflowStateOverride.loops || {}
parallels = metadata.workflowStateOverride.parallels || {}
logger.info(`[${requestId}] Using workflow state override (diff workflow execution)`, {
blocksCount: Object.keys(blocks).length,
edgesCount: edges.length,
})
} else if (useDraftState) {
const draftData = await loadWorkflowFromNormalizedTables(workflowId)
if (!draftData) {