improvement(wand): upgrade wand to use SSE (#1100)

* improvement(wand): upgrade wand to use SSE

* fix(ocr-azure): added OCR_AZURE_API_KEY envvar (#1102)

* make wand identical to chat panel

Author: waleedlatif1

apps/sim/app/api/wand-generate/route.ts

@@ -95,12 +95,19 @@ export async function POST(req: NextRequest) {
       {
         stream,
         historyLength: history.length,
+        endpoint: useWandAzure ? azureEndpoint : 'api.openai.com',
+        model: useWandAzure ? wandModelName : 'gpt-4o',
+        apiVersion: useWandAzure ? azureApiVersion : 'N/A',
       }
     )
 
     // For streaming responses
     if (stream) {
       try {
+        logger.debug(
+          `[${requestId}] Starting streaming request to ${useWandAzure ? 'Azure OpenAI' : 'OpenAI'}`
+        )
+
         const streamCompletion = await client.chat.completions.create({
           model: useWandAzure ? wandModelName : 'gpt-4o',
           messages: messages,
@@ -109,6 +116,8 @@ export async function POST(req: NextRequest) {
           stream: true,
         })
 
+        logger.debug(`[${requestId}] Stream connection established successfully`)
+
         return new Response(
           new ReadableStream({
             async start(controller) {
@@ -118,31 +127,34 @@ export async function POST(req: NextRequest) {
                 for await (const chunk of streamCompletion) {
                   const content = chunk.choices[0]?.delta?.content || ''
                   if (content) {
-                    // Use the same format as codegen API for consistency
+                    // Use SSE format identical to chat streaming
                     controller.enqueue(
-                      encoder.encode(`${JSON.stringify({ chunk: content, done: false })}\n`)
+                      encoder.encode(`data: ${JSON.stringify({ chunk: content })}\n\n`)
                     )
                   }
                 }
 
-                // Send completion signal
-                controller.enqueue(encoder.encode(`${JSON.stringify({ chunk: '', done: true })}\n`))
+                // Send completion signal in SSE format
+                controller.enqueue(encoder.encode(`data: ${JSON.stringify({ done: true })}\n\n`))
                 controller.close()
                 logger.info(`[${requestId}] Wand generation streaming completed`)
               } catch (streamError: any) {
                 logger.error(`[${requestId}] Streaming error`, { error: streamError.message })
                 controller.enqueue(
-                  encoder.encode(`${JSON.stringify({ error: 'Streaming failed', done: true })}\n`)
+                  encoder.encode(
+                    `data: ${JSON.stringify({ error: 'Streaming failed', done: true })}\n\n`
+                  )
                 )
                 controller.close()
               }
             },
           }),
           {
             headers: {
-              'Content-Type': 'text/plain',
-              'Cache-Control': 'no-cache, no-transform',
+              'Content-Type': 'text/event-stream',
+              'Cache-Control': 'no-cache',
               Connection: 'keep-alive',
+              'X-Accel-Buffering': 'no',
             },
           }
         )

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-wand.ts

@@ -198,35 +198,37 @@ export function useWand({
             const { done, value } = await reader.read()
             if (done) break
 
-            // Process incoming chunks
-            const text = decoder.decode(value)
-            const lines = text.split('\n').filter((line) => line.trim() !== '')
+            // Process incoming chunks using SSE format (identical to Chat panel)
+            const chunk = decoder.decode(value)
+            const lines = chunk.split('\n\n')
 
             for (const line of lines) {
-              try {
-                const data = JSON.parse(line)
+              if (line.startsWith('data: ')) {
+                try {
+                  const data = JSON.parse(line.substring(6))
 
-                // Check if there's an error
-                if (data.error) {
-                  throw new Error(data.error)
-                }
+                  // Check if there's an error
+                  if (data.error) {
+                    throw new Error(data.error)
+                  }
 
-                // Process chunk
-                if (data.chunk && !data.done) {
-                  accumulatedContent += data.chunk
-                  // Stream each chunk to the UI immediately
-                  if (onStreamChunk) {
-                    onStreamChunk(data.chunk)
+                  // Process chunk
+                  if (data.chunk) {
+                    accumulatedContent += data.chunk
+                    // Stream each chunk to the UI immediately
+                    if (onStreamChunk) {
+                      onStreamChunk(data.chunk)
+                    }
                   }
-                }
 
-                // Check if streaming is complete
-                if (data.done) {
-                  break
+                  // Check if streaming is complete
+                  if (data.done) {
+                    break
+                  }
+                } catch (parseError) {
+                  // Continue processing other lines
+                  logger.debug('Failed to parse SSE line', { line, parseError })
                 }
-              } catch (parseError) {
-                // Continue processing other lines
-                logger.debug('Failed to parse streaming line', { line, parseError })
               }
             }
           }