change app.py

app.py

@@ -3,6 +3,7 @@ import asyncio
 import base64
 import json
 import os
+import queue
 import threading
 import time
 import uuid
@@ -124,7 +125,8 @@ class UserSession:
     """Per-user session state."""
     def __init__(self):
         self.session_id = str(uuid.uuid4())
-        self._audio_queue = None  # Created lazily in the correct event loop
+        # Use a thread-safe queue for cross-thread communication
+        self._audio_queue = queue.Queue(maxsize=200)
         self.transcription_text = ""
         self.is_running = False
         self.status_message = "ready"
@@ -137,14 +139,12 @@ class UserSession:
     
     @property
     def audio_queue(self):
-        """Lazily create audio queue to ensure it's in the right event loop."""
-        if self._audio_queue is None:
-            self._audio_queue = asyncio.Queue(maxsize=100)
+        """Return the thread-safe queue."""
         return self._audio_queue
     
     def reset_queue(self):
         """Reset the audio queue."""
-        self._audio_queue = asyncio.Queue(maxsize=100)
+        self._audio_queue = queue.Queue(maxsize=200)
 
 
 # Load CSS from external file
@@ -298,8 +298,12 @@ async def send_silence(ws, duration=2.0):
 async def websocket_handler(session):
     """Connect to WebSocket and handle audio streaming + transcription."""
     try:
-        async with websockets.connect(ws_url) as ws:
-            await ws.recv()
+        # Add connection timeout to prevent hanging
+        async with asyncio.timeout(10):  # 10 second connection timeout
+            ws = await websockets.connect(ws_url)
+        
+        async with ws:
+            await asyncio.wait_for(ws.recv(), timeout=5)
             await ws.send(json.dumps({"type": "session.update", "model": model}))
 
             session.status_message = "warming"
@@ -325,15 +329,18 @@ async def websocket_handler(session):
                                 session.status_message = "timeout"
                                 break
                         
+                        # Use thread-safe queue with non-blocking get + async sleep
                         try:
-                            chunk = await asyncio.wait_for(session.audio_queue.get(), timeout=0.1)
+                            chunk = session.audio_queue.get_nowait()
                             if session.is_running:
                                 await ws.send(
                                     json.dumps(
                                         {"type": "input_audio_buffer.append", "audio": chunk}
                                     )
                                 )
-                        except asyncio.TimeoutError:
+                        except queue.Empty:
+                            # No audio available, yield control briefly
+                            await asyncio.sleep(0.05)
                             continue
                     except Exception as e:
                         if session.is_running:  # Only log if unexpected
@@ -376,6 +383,9 @@ async def websocket_handler(session):
         pass  # Normal cancellation
     except websockets.exceptions.ConnectionClosed:
         pass  # Normal closure
+    except asyncio.TimeoutError:
+        print(f"WebSocket connection timeout for session {session.session_id[:8]}")
+        session.status_message = "error"
     except Exception as e:
         error_msg = str(e) if str(e) else type(e).__name__
         if "ConnectionReset" not in error_msg:  # Suppress common disconnect errors
@@ -386,7 +396,8 @@ async def websocket_handler(session):
         # Remove from active sessions
         with _sessions_lock:
             _active_sessions.pop(session.session_id, None)
-        # Note: Don't remove from registry here - session might be reused
+            active_count = len(_active_sessions)
+        print(f"Session {session.session_id[:8]} ended. Active sessions: {active_count}")
 
 
 def start_websocket(session):
@@ -396,6 +407,9 @@ def start_websocket(session):
     # Register this session
     with _sessions_lock:
         _active_sessions[session.session_id] = session
+        active_count = len(_active_sessions)
+    
+    print(f"Starting session {session.session_id[:8]}. Active sessions: {active_count}")
     
     # Submit to the shared event loop
     loop = get_event_loop()
@@ -538,12 +552,11 @@ def process_audio(audio, session_id):
         pcm16 = (audio_float * 32767).astype(np.int16)
         b64_chunk = base64.b64encode(pcm16.tobytes()).decode("utf-8")
         
-        # Non-blocking put to async queue (thread-safe)
+        # Put directly into thread-safe queue (no event loop needed)
         try:
-            loop = get_event_loop()
-            loop.call_soon_threadsafe(lambda: _safe_queue_put(session.audio_queue, b64_chunk))
+            session.audio_queue.put_nowait(b64_chunk)
         except Exception:
-            pass  # Skip if queue is full or loop issues
+            pass  # Skip if queue is full
 
         return get_transcription_html(session.transcription_text, session.status_message, session.current_wpm), current_session_id
     except Exception as e:
@@ -552,12 +565,6 @@ def process_audio(audio, session_id):
         return get_transcription_html("", "error", ""), current_session_id
 
 
-def _safe_queue_put(q, item):
-    """Safely put item in async queue without blocking."""
-    try:
-        q.put_nowait(item)
-    except asyncio.QueueFull:
-        pass  # Drop frame if queue is full
 
 
 # Gradio interface