fix(execution): type event buffer with ExecutionEvent instead of Record<string, unknown>

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -720,7 +720,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
             }
           }
           if (event.type !== 'stream:chunk' && event.type !== 'stream:done') {
-            eventWriter.write(event as unknown as Record<string, unknown>).catch(() => {})
+            eventWriter.write(event).catch(() => {})
           }
         }
 

apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts

@@ -7,7 +7,7 @@ import {
   getExecutionMeta,
   readExecutionEvents,
 } from '@/lib/execution/event-buffer'
-import { type ExecutionEvent, formatSSEEvent } from '@/lib/workflows/executor/execution-events'
+import { formatSSEEvent } from '@/lib/workflows/executor/execution-events'
 import { authorizeWorkflowByWorkspacePermission } from '@/lib/workflows/utils'
 
 const logger = createLogger('ExecutionStreamReconnectAPI')
@@ -88,7 +88,7 @@ export async function GET(
           const events = await readExecutionEvents(executionId, lastEventId)
           for (const entry of events) {
             if (closed) return
-            enqueue(formatSSEEvent(entry.event as unknown as ExecutionEvent))
+            enqueue(formatSSEEvent(entry.event))
             lastEventId = entry.eventId
           }
 
@@ -108,7 +108,7 @@ export async function GET(
             const newEvents = await readExecutionEvents(executionId, lastEventId)
             for (const entry of newEvents) {
               if (closed) return
-              enqueue(formatSSEEvent(entry.event as unknown as ExecutionEvent))
+              enqueue(formatSSEEvent(entry.event))
               lastEventId = entry.eventId
             }
 
@@ -118,7 +118,7 @@ export async function GET(
               const finalEvents = await readExecutionEvents(executionId, lastEventId)
               for (const entry of finalEvents) {
                 if (closed) return
-                enqueue(formatSSEEvent(entry.event as unknown as ExecutionEvent))
+                enqueue(formatSSEEvent(entry.event))
                 lastEventId = entry.eventId
               }
               enqueue('data: [DONE]\n\n')

apps/sim/lib/execution/event-buffer.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { getRedisClient } from '@/lib/core/config/redis'
+import type { ExecutionEvent } from '@/lib/workflows/executor/execution-events'
 
 const logger = createLogger('ExecutionEventBuffer')
 
@@ -34,11 +35,11 @@ export interface ExecutionStreamMeta {
 export interface ExecutionEventEntry {
   eventId: number
   executionId: string
-  event: Record<string, unknown>
+  event: ExecutionEvent
 }
 
 export interface ExecutionEventWriter {
-  write: (event: Record<string, unknown>) => Promise<ExecutionEventEntry>
+  write: (event: ExecutionEvent) => Promise<ExecutionEventEntry>
   flush: () => Promise<void>
   close: () => Promise<void>
 }
@@ -199,7 +200,7 @@ export function createExecutionEventWriter(executionId: string): ExecutionEventW
     }
   }
 
-  const write = async (event: Record<string, unknown>) => {
+  const write = async (event: ExecutionEvent) => {
     if (closed) return { eventId: 0, executionId, event }
     if (nextEventId === 0 || nextEventId > maxReservedId) {
       await reserveIds(1)