Use one SSE stream

project/chat.py

@@ -223,16 +223,16 @@ def create_variable(var_name):
         traceback.print_exc()
         return f"Error creating variable: {str(e)}"
 
-# Server-Sent Events endpoint for creation requests
-@app.get("/create_stream")
-async def create_stream():
-    """Stream creation requests to the frontend using Server-Sent Events"""
+# Unified Server-Sent Events endpoint for all workspace operations
+@app.get("/unified_stream")
+async def unified_stream():
+    """Unified SSE endpoint for delete, create, and variable operations"""
     
-    async def clear_sent_request(sent_requests, request_id, delay):
-        """Clear request_id from sent_requests after delay seconds"""
+    async def clear_sent_request(sent_requests, request_key, delay):
+        """Clear request_key from sent_requests after delay seconds"""
         await asyncio.sleep(delay)
-        if request_id in sent_requests:
-            sent_requests.discard(request_id)
+        if request_key in sent_requests:
+            sent_requests.discard(request_key)
     
     async def event_generator():
         sent_requests = set()  # Track sent requests to avoid duplicates
@@ -240,23 +240,60 @@ async def create_stream():
         
         while True:
             try:
-                # Check for creation requests (non-blocking)
-                if not creation_queue.empty():
+                # Check deletion queue
+                if not deletion_queue.empty():
+                    deletion_request = deletion_queue.get_nowait()
+                    block_id = deletion_request.get("block_id")
+                    request_key = f"delete_{block_id}"
+                    
+                    # Avoid sending duplicate requests too quickly
+                    if request_key not in sent_requests:
+                        sent_requests.add(request_key)
+                        deletion_request["type"] = "delete"  # Add type identifier
+                        print(f"[SSE SEND] Sending deletion request for block: {block_id}")
+                        yield f"data: {json.dumps(deletion_request)}\n\n"
+                        
+                        # Clear from sent_requests after 10 seconds
+                        asyncio.create_task(clear_sent_request(sent_requests, request_key, 10))
+                    else:
+                        print(f"[SSE SKIP] Skipping duplicate request for block: {block_id}")
+                
+                # Check creation queue
+                elif not creation_queue.empty():
                     creation_request = creation_queue.get_nowait()
                     request_id = creation_request.get("request_id")
+                    request_key = f"create_{request_id}"
                     
                     # Avoid sending duplicate requests too quickly
-                    if request_id not in sent_requests:
-                        sent_requests.add(request_id)
-                        print(f"[SSE CREATE SEND] Sending creation request with ID: {request_id}")
+                    if request_key not in sent_requests:
+                        sent_requests.add(request_key)
+                        creation_request["type"] = "create"  # Add type identifier
+                        print(f"[SSE SEND] Sending creation request with ID: {request_id}")
                         yield f"data: {json.dumps(creation_request)}\n\n"
                         
                         # Clear from sent_requests after 10 seconds
-                        asyncio.create_task(clear_sent_request(sent_requests, request_id, 10))
+                        asyncio.create_task(clear_sent_request(sent_requests, request_key, 10))
                     else:
-                        print(f"[SSE CREATE SKIP] Skipping duplicate request for ID: {request_id}")
+                        print(f"[SSE SKIP] Skipping duplicate request for ID: {request_id}")
+                
+                # Check variable queue
+                elif not variable_queue.empty():
+                    variable_request = variable_queue.get_nowait()
+                    request_id = variable_request.get("request_id")
+                    request_key = f"variable_{request_id}"
                     
-                    await asyncio.sleep(0.1)  # Small delay between messages
+                    # Avoid sending duplicate requests too quickly
+                    if request_key not in sent_requests:
+                        sent_requests.add(request_key)
+                        variable_request["type"] = "variable"  # Add type identifier
+                        print(f"[SSE SEND] Sending variable creation request with ID: {request_id}")
+                        yield f"data: {json.dumps(variable_request)}\n\n"
+                        
+                        # Clear from sent_requests after 10 seconds
+                        asyncio.create_task(clear_sent_request(sent_requests, request_key, 10))
+                    else:
+                        print(f"[SSE SKIP] Skipping duplicate request for ID: {request_id}")
+                
                 else:
                     # Send a heartbeat every 30 seconds to keep connection alive
                     heartbeat_counter += 1
@@ -264,10 +301,11 @@ async def create_stream():
                         yield f"data: {json.dumps({'heartbeat': True})}\n\n"
                         heartbeat_counter = 0
                     await asyncio.sleep(0.1)
+                    
             except queue.Empty:
                 await asyncio.sleep(0.1)
             except Exception as e:
-                print(f"[SSE CREATE ERROR] {e}")
+                print(f"[SSE ERROR] {e}")
                 await asyncio.sleep(1)
     
     return StreamingResponse(
@@ -299,63 +337,6 @@ async def creation_result(request: Request):
     
     return {"received": True}
 
-# Server-Sent Events endpoint for deletion requests
-@app.get("/delete_stream")
-async def delete_stream():
-    """Stream deletion requests to the frontend using Server-Sent Events"""
-    
-    async def clear_sent_request(sent_requests, block_id, delay):
-        """Clear block_id from sent_requests after delay seconds"""
-        await asyncio.sleep(delay)
-        if block_id in sent_requests:
-            sent_requests.discard(block_id)
-    
-    async def event_generator():
-        sent_requests = set()  # Track sent requests to avoid duplicates
-        heartbeat_counter = 0
-        
-        while True:
-            try:
-                # Check for deletion requests (non-blocking)
-                if not deletion_queue.empty():
-                    deletion_request = deletion_queue.get_nowait()
-                    block_id = deletion_request.get("block_id")
-                    
-                    # Avoid sending duplicate requests too quickly
-                    if block_id not in sent_requests:
-                        sent_requests.add(block_id)
-                        print(f"[SSE SEND] Sending deletion request for block: {block_id}")
-                        yield f"data: {json.dumps(deletion_request)}\n\n"
-                        
-                        # Clear from sent_requests after 10 seconds
-                        asyncio.create_task(clear_sent_request(sent_requests, block_id, 10))
-                    else:
-                        print(f"[SSE SKIP] Skipping duplicate request for block: {block_id}")
-                    
-                    await asyncio.sleep(0.1)  # Small delay between messages
-                else:
-                    # Send a heartbeat every 30 seconds to keep connection alive
-                    heartbeat_counter += 1
-                    if heartbeat_counter >= 300:  # 300 * 0.1 = 30 seconds
-                        yield f"data: {json.dumps({'heartbeat': True})}\n\n"
-                        heartbeat_counter = 0
-                    await asyncio.sleep(0.1)
-            except queue.Empty:
-                await asyncio.sleep(0.1)
-            except Exception as e:
-                print(f"[SSE ERROR] {e}")
-                await asyncio.sleep(1)
-    
-    return StreamingResponse(
-        event_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-        }
-    )
-
 # Endpoint to receive deletion results from frontend
 @app.post("/deletion_result")
 async def deletion_result(request: Request):
@@ -374,64 +355,6 @@ async def deletion_result(request: Request):
     
     return {"received": True}
 
-# Server-Sent Events endpoint for variable creation requests
-@app.get("/variable_stream")
-async def variable_stream():
-    """Stream variable creation requests to the frontend using Server-Sent Events"""
-    
-    async def clear_sent_request(sent_requests, request_id, delay):
-        """Clear request_id from sent_requests after delay seconds"""
-        await asyncio.sleep(delay)
-        if request_id in sent_requests:
-            sent_requests.discard(request_id)
-    
-    async def event_generator():
-        sent_requests = set()  # Track sent requests to avoid duplicates
-        heartbeat_counter = 0
-        
-        while True:
-            try:
-                # Check for variable creation requests (non-blocking)
-                if not variable_queue.empty():
-                    var_request = variable_queue.get_nowait()
-                    request_id = var_request.get("request_id")
-                    
-                    # Avoid sending duplicate requests too quickly
-                    if request_id not in sent_requests:
-                        sent_requests.add(request_id)
-                        print(f"[SSE VARIABLE SEND] Sending variable creation request with ID: {request_id}")
-                        yield f"data: {json.dumps(var_request)}\n\n"
-                        
-                        # Clear from sent_requests after 10 seconds
-                        asyncio.create_task(clear_sent_request(sent_requests, request_id, 10))
-                    else:
-                        print(f"[SSE VARIABLE SKIP] Skipping duplicate request for ID: {request_id}")
-                    
-                    await asyncio.sleep(0.1)  # Small delay between messages
-                else:
-                    # Send a heartbeat every 30 seconds to keep connection alive
-                    heartbeat_counter += 1
-                    if heartbeat_counter >= 300:  # 300 * 0.1 = 30 seconds
-                        yield f"data: {json.dumps({'heartbeat': True})}\n\n"
-                        heartbeat_counter = 0
-                    await asyncio.sleep(0.1)
-                    
-            except queue.Empty:
-                await asyncio.sleep(0.1)
-            except Exception as e:
-                print(f"[SSE VARIABLE ERROR] {e}")
-                await asyncio.sleep(1)
-    
-    return StreamingResponse(
-        event_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-        }
-    )
-
 # Endpoint to receive variable creation results from frontend
 @app.post("/variable_result")
 async def variable_result(request: Request):

project/src/index.js

@@ -223,10 +223,10 @@ cleanWorkspace.addEventListener("click", () => {
   ws.cleanUp();
 });
 
-// Set up SSE connection for deletion requests
-const setupDeletionStream = () => {
-  const eventSource = new EventSource('/delete_stream');
-  const processedRequests = new Set(); // Track processed deletion requests
+// Set up unified SSE connection for all workspace operations
+const setupUnifiedStream = () => {
+  const eventSource = new EventSource('/unified_stream');
+  const processedRequests = new Set(); // Track processed requests
 
   eventSource.onmessage = (event) => {
     try {
@@ -235,19 +235,29 @@ const setupDeletionStream = () => {
       // Skip heartbeat messages
       if (data.heartbeat) return;
 
-      // Skip if we've already processed this exact request
-      const requestKey = `${data.block_id}_${Date.now()}`;
-      if (data.block_id && processedRequests.has(data.block_id)) {
-        console.log('[SSE] Skipping duplicate deletion request for:', data.block_id);
+      // Determine request key based on type
+      let requestKey;
+      if (data.type === 'delete') {
+        requestKey = `delete_${data.block_id}`;
+      } else if (data.type === 'create') {
+        requestKey = `create_${data.request_id}`;
+      } else if (data.type === 'variable') {
+        requestKey = `variable_${data.request_id}`;
+      }
+
+      // Skip if we've already processed this request
+      if (requestKey && processedRequests.has(requestKey)) {
+        console.log('[SSE] Skipping duplicate request:', requestKey);
         return;
       }
-      if (data.block_id) {
-        processedRequests.add(data.block_id);
+      if (requestKey) {
+        processedRequests.add(requestKey);
         // Clear after 10 seconds to allow retries if needed
-        setTimeout(() => processedRequests.delete(data.block_id), 10000);
+        setTimeout(() => processedRequests.delete(requestKey), 10000);
       }
 
-      if (data.block_id) {
+      // Handle deletion requests
+      if (data.type === 'delete' && data.block_id) {
         console.log('[SSE] Received deletion request for block:', data.block_id);
 
         // Try to delete the block
@@ -292,53 +302,9 @@ const setupDeletionStream = () => {
           console.error('[SSE] Error sending deletion result:', err);
         });
       }
-    } catch (err) {
-      console.error('[SSE] Error processing message:', err);
-    }
-  };
-
-  eventSource.onerror = (error) => {
-    console.error('[SSE] Connection error:', error);
-    // Reconnect after 5 seconds
-    setTimeout(() => {
-      console.log('[SSE] Attempting to reconnect...');
-      setupDeletionStream();
-    }, 5000);
-  };
-
-  eventSource.onopen = () => {
-    console.log('[SSE] Connected to deletion stream');
-  };
-};
-
-// Start the SSE connection
-setupDeletionStream();
-
-// Set up SSE connection for creation requests
-const setupCreationStream = () => {
-  const eventSource = new EventSource('/create_stream');
-  const processedRequests = new Set(); // Track processed creation requests
-
-  eventSource.onmessage = (event) => {
-    try {
-      const data = JSON.parse(event.data);
-
-      // Skip heartbeat messages
-      if (data.heartbeat) return;
-
-      // Skip if we've already processed this request
-      if (data.request_id && processedRequests.has(data.request_id)) {
-        console.log('[SSE CREATE] Skipping duplicate creation request:', data.request_id);
-        return;
-      }
-      if (data.request_id) {
-        processedRequests.add(data.request_id);
-        // Clear after 10 seconds to allow retries if needed
-        setTimeout(() => processedRequests.delete(data.request_id), 10000);
-      }
-
-      if (data.block_spec && data.request_id) {
-        console.log('[SSE CREATE] Received creation request:', data.request_id, data.block_spec);
+      // Handle creation requests
+      else if (data.type === 'create' && data.block_spec && data.request_id) {
+        console.log('[SSE] Received creation request:', data.request_id, data.block_spec);
 
         let success = false;
         let error = null;
@@ -705,52 +671,9 @@ const setupCreationStream = () => {
           console.error('[SSE CREATE] Error sending creation result:', err);
         });
       }
-    } catch (err) {
-      console.error('[SSE CREATE] Error processing message:', err);
-    }
-  };
-
-  eventSource.onerror = (error) => {
-    console.error('[SSE CREATE] Connection error:', error);
-    // Reconnect after 5 seconds
-    setTimeout(() => {
-      console.log('[SSE CREATE] Attempting to reconnect...');
-      setupCreationStream();
-    }, 5000);
-  };
-
-  eventSource.onopen = () => {
-    console.log('[SSE CREATE] Connected to creation stream');
-  };
-};
-
-// Start the creation SSE connection
-setupCreationStream();
-
-const setupVariableStream = () => {
-  const eventSource = new EventSource('/variable_stream');
-  const processedRequests = new Set(); // Track processed variable requests
-
-  eventSource.onmessage = (event) => {
-    try {
-      const data = JSON.parse(event.data);
-
-      // Skip heartbeat messages
-      if (data.heartbeat) return;
-
-      // Skip if we've already processed this request
-      if (data.request_id && processedRequests.has(data.request_id)) {
-        console.log('[SSE VARIABLE] Skipping duplicate variable request:', data.request_id);
-        return;
-      }
-      if (data.request_id) {
-        processedRequests.add(data.request_id);
-        // Clear after 10 seconds to allow retries if needed
-        setTimeout(() => processedRequests.delete(data.request_id), 10000);
-      }
-
-      if (data.variable_name && data.request_id) {
-        console.log('[SSE VARIABLE] Received variable creation request:', data.request_id, data.variable_name);
+      // Handle variable creation requests
+      else if (data.type === 'variable' && data.variable_name && data.request_id) {
+        console.log('[SSE] Received variable creation request:', data.request_id, data.variable_name);
 
         let success = false;
         let error = null;
@@ -766,18 +689,18 @@ const setupVariableStream = () => {
           if (variableModel) {
             variableId = variableModel.getId();
             success = true;
-            console.log('[SSE VARIABLE] Successfully created variable:', variableName, 'with ID:', variableId);
+            console.log('[SSE] Successfully created variable:', variableName, 'with ID:', variableId);
           } else {
             throw new Error('Failed to create variable model');
           }
 
         } catch (e) {
           error = e.toString();
-          console.error('[SSE VARIABLE] Error creating variable:', e);
+          console.error('[SSE] Error creating variable:', e);
         }
 
         // Send result back to backend immediately
-        console.log('[SSE VARIABLE] Sending variable creation result:', {
+        console.log('[SSE] Sending variable creation result:', {
           request_id: data.request_id,
           success,
           error,
@@ -794,32 +717,32 @@ const setupVariableStream = () => {
             variable_id: variableId
           })
         }).then(response => {
-          console.log('[SSE VARIABLE] Variable creation result sent successfully');
+          console.log('[SSE] Variable creation result sent successfully');
         }).catch(err => {
-          console.error('[SSE VARIABLE] Error sending variable creation result:', err);
+          console.error('[SSE] Error sending variable creation result:', err);
         });
       }
     } catch (err) {
-      console.error('[SSE VARIABLE] Error processing message:', err);
+      console.error('[SSE] Error processing message:', err);
     }
   };
 
   eventSource.onerror = (error) => {
-    console.error('[SSE VARIABLE] Connection error:', error);
+    console.error('[SSE] Connection error:', error);
     // Reconnect after 5 seconds
     setTimeout(() => {
-      console.log('[SSE VARIABLE] Attempting to reconnect...');
-      setupVariableStream();
+      console.log('[SSE] Attempting to reconnect...');
+      setupUnifiedStream();
     }, 5000);
   };
 
   eventSource.onopen = () => {
-    console.log('[SSE VARIABLE] Connected to variable stream');
+    console.log('[SSE] Connected to unified stream');
   };
 };
 
-// Start the variable SSE connection
-setupVariableStream();
+// Start the unified SSE connection
+setupUnifiedStream();
 
 // Observe any size change to the blockly container
 const observer = new ResizeObserver(() => {

project/unified_server.py

@@ -34,26 +34,18 @@ async def update_chat_route(request: Request):
 async def set_api_key_chat_route(request: Request):
     return await chat.set_api_key_chat(request)
 
-@app.get("/create_stream")
-async def create_stream_route():
-    return await chat.create_stream()
+@app.get("/unified_stream")
+async def unified_stream_route():
+    return await chat.unified_stream()
 
 @app.post("/creation_result")
 async def creation_result_route(request: Request):
     return await chat.creation_result(request)
 
-@app.get("/delete_stream")
-async def delete_stream_route():
-    return await chat.delete_stream()
-
 @app.post("/deletion_result")
 async def deletion_result_route(request: Request):
     return await chat.deletion_result(request)
 
-@app.get("/variable_stream")
-async def variable_stream_route():
-    return await chat.variable_stream()
-
 @app.post("/variable_result")
 async def variable_result_route(request: Request):
     return await chat.variable_result(request)