From 4c2764f09c0a38418b79402bdbbc71e80c5cfb13 Mon Sep 17 00:00:00 2001
From: Emir Karabeg <emirkarabeg@berkeley.edu>
Date: Sat, 15 Feb 2025 20:01:55 -0800
Subject: [PATCH] feature(console): stream console messages back from executor

---
 app/w/hooks/use-workflow-execution.ts | 53 +++------------------------
 executor/index.ts                     | 32 +++++++++++++++-
 2 files changed, 36 insertions(+), 49 deletions(-)

diff --git a/app/w/hooks/use-workflow-execution.ts b/app/w/hooks/use-workflow-execution.ts
index 816213993d..685d0b5649 100644
--- a/app/w/hooks/use-workflow-execution.ts
+++ b/app/w/hooks/use-workflow-execution.ts
@@ -14,7 +14,7 @@ export function useWorkflowExecution() {
   const { blocks, edges, loops } = useWorkflowStore()
   const { activeWorkflowId } = useWorkflowRegistry()
   const { addNotification } = useNotificationStore()
-  const { addConsole, toggleConsole, isOpen } = useConsoleStore()
+  const { toggleConsole, isOpen } = useConsoleStore()
   const { getAllVariables } = useEnvironmentStore()
 
   const handleRunWorkflow = useCallback(async () => {
@@ -52,42 +52,11 @@ export function useWorkflowExecution() {
       // Execute workflow
       const workflow = new Serializer().serializeWorkflow(blocks, edges, loops)
       const executor = new Executor(workflow, currentBlockStates, envVarValues)
+      const result = await executor.execute(activeWorkflowId)
 
-      const result = await executor.execute('my-run-id')
       setExecutionResult(result)
 
-      // Add console entries for each block execution
-      if (result.logs) {
-        result.logs.forEach((log) => {
-          addConsole({
-            output: log.output,
-            error: log.error,
-            durationMs: log.durationMs,
-            startedAt: log.startedAt,
-            endedAt: log.endedAt,
-            workflowId: activeWorkflowId,
-            timestamp: log.startedAt,
-            blockName: log.blockName,
-            blockType: log.blockType,
-          })
-        })
-      }
-
-      if (result.logs) {
-        console.group('Detailed Block Logs')
-        result.logs.forEach((log) => {
-          console.log(`Block ${log.blockName}: Success=${log.success}`, {
-            output: log.output,
-            error: log.error,
-            durationMs: log.durationMs,
-            startedAt: log.startedAt,
-            endedAt: log.endedAt,
-          })
-        })
-        console.groupEnd()
-      }
-
-      // Show execution result with workflowId
+      // Show execution result notification
       addNotification(
         result.success ? 'console' : 'error',
         result.success
@@ -95,7 +64,7 @@ export function useWorkflowExecution() {
           : `Workflow execution failed: ${result.error}`,
         activeWorkflowId
       )
-    } catch (error) {
+    } catch (error: any) {
       const errorMessage = error instanceof Error ? error.message : 'Unknown error'
       setExecutionResult({
         success: false,
@@ -104,18 +73,6 @@ export function useWorkflowExecution() {
         logs: [],
       })
 
-      // Add error entry to console
-      addConsole({
-        output: {},
-        error: errorMessage,
-        durationMs: -1,
-        startedAt: new Date().toISOString(),
-        endedAt: new Date().toISOString(),
-        workflowId: activeWorkflowId,
-        timestamp: new Date().toISOString(),
-        blockName: 'Error',
-      })
-
       addNotification('error', `Workflow execution failed: ${errorMessage}`, activeWorkflowId)
     } finally {
       setIsExecuting(false)
@@ -124,8 +81,8 @@ export function useWorkflowExecution() {
     activeWorkflowId,
     blocks,
     edges,
+    loops,
     addNotification,
-    addConsole,
     isOpen,
     toggleConsole,
     getAllVariables,
diff --git a/executor/index.ts b/executor/index.ts
index 71f182e36e..70fceb9fa6 100644
--- a/executor/index.ts
+++ b/executor/index.ts
@@ -1,3 +1,4 @@
+import { useConsoleStore } from '@/stores/console/store'
 import { getAllBlocks } from '@/blocks'
 import { generateRouterPrompt } from '@/blocks/blocks/router'
 import { BlockOutput } from '@/blocks/types'
@@ -433,6 +434,7 @@ export class Executor {
     }
 
     const blockLog = this.startBlockLog(block)
+    const addConsole = useConsoleStore.getState().addConsole
 
     try {
       let output: BlockOutput
@@ -579,18 +581,46 @@ export class Executor {
         output = { response: result.output }
       }
 
+      // Log success
       blockLog.success = true
       blockLog.output = output
       this.finalizeBlockLog(blockLog)
       context.blockLogs.push(blockLog)
 
+      // Add to console immediately
+      addConsole({
+        output: blockLog.output,
+        durationMs: blockLog.durationMs,
+        startedAt: blockLog.startedAt,
+        endedAt: blockLog.endedAt,
+        workflowId: context.workflowId,
+        timestamp: blockLog.startedAt,
+        blockName: block.metadata?.name || 'Unnamed Block',
+        blockType: block.metadata?.id || 'unknown',
+      })
+
       context.blockStates.set(block.id, output)
       return output
     } catch (error: any) {
+      // Log error
       blockLog.success = false
-      blockLog.error = error.message || 'Block execution failed'
+      blockLog.error = error.message
       this.finalizeBlockLog(blockLog)
       context.blockLogs.push(blockLog)
+
+      // Add error to console immediately
+      addConsole({
+        output: {},
+        error: error.message,
+        durationMs: blockLog.durationMs,
+        startedAt: blockLog.startedAt,
+        endedAt: blockLog.endedAt,
+        workflowId: context.workflowId,
+        timestamp: blockLog.startedAt,
+        blockName: block.metadata?.name || 'Unnamed Block',
+        blockType: block.metadata?.id || 'unknown',
+      })
+
       throw error
     }
   }