Fix: Use user-specific rooms instead of socket ID tracking for event isolation

Co-authored-by: lstein <111189+lstein@users.noreply.github.com>

Author: Copilot

invokeai/app/api/sockets.py

@@ -159,7 +159,7 @@ async def _handle_disconnect(self, sid: str) -> None:
             logger.debug(f"Socket {sid} disconnected and cleaned up")
 
     async def _handle_sub_queue(self, sid: str, data: Any) -> None:
-        """Handle queue subscription and ensure user info is available for this socket."""
+        """Handle queue subscription and add socket to both queue and user-specific rooms."""
         queue_id = QueueSubscriptionEvent(**data).queue_id
 
         # Check if we have user info for this socket
@@ -171,9 +171,21 @@ async def _handle_sub_queue(self, sid: str, data: Any) -> None:
                 "is_admin": False,
             }
 
-        await self._sio.enter_room(sid, queue_id)
         user_id = self._socket_users[sid]['user_id']
-        logger.info(f"Socket {sid} (user_id: {user_id}) subscribed to queue {queue_id}")
+        is_admin = self._socket_users[sid]['is_admin']
+        
+        # Add socket to the queue room
+        await self._sio.enter_room(sid, queue_id)
+        
+        # Also add socket to a user-specific room for event filtering
+        user_room = f"user:{user_id}"
+        await self._sio.enter_room(sid, user_room)
+        
+        # If admin, also add to admin room to receive all events
+        if is_admin:
+            await self._sio.enter_room(sid, "admin")
+        
+        logger.info(f"Socket {sid} (user_id: {user_id}, is_admin: {is_admin}) subscribed to queue {queue_id} and user room {user_room}")
 
     async def _handle_unsub_queue(self, sid: str, data: Any) -> None:
         await self._sio.leave_room(sid, QueueSubscriptionEvent(**data).queue_id)
@@ -195,45 +207,24 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
 
             # Check if this is a queue item event that should be filtered by user
             if isinstance(event_data, QueueItemEventBase) and hasattr(event_data, "user_id"):
-                # Get all socket IDs in the queue room (this is a synchronous generator, not awaitable)
-                room_name = event_data.queue_id
-                room_participants = self._sio.manager.get_participants("/", room_name)
+                # Emit to user-specific room and admin room
+                user_room = f"user:{event_data.user_id}"
 
-                # Filter sids based on user_id or admin status
-                emitted_count = 0
-                participants_checked = 0
-                for participant in room_participants:
-                    participants_checked += 1
-                    # get_participants returns tuples of (namespace, sid) - extract the sid
-                    sid = participant[1] if isinstance(participant, tuple) else participant
-                    try:
-                        # Get user info from internal dict instead of socket session
-                        user_info = self._socket_users.get(sid)
-                        if user_info:
-                            session_user_id = user_info.get("user_id", "system")
-                            is_admin = user_info.get("is_admin", False)
-                            
-                            logger.debug(f"Checking socket {sid}: session_user_id={session_user_id}, event_user_id={event_data.user_id}, is_admin={is_admin}")
-                            
-                            # Emit to the owner or to admins
-                            if session_user_id == event_data.user_id or is_admin:
-                                await self._sio.emit(
-                                    event=event_name,
-                                    data=event_data.model_dump(mode="json"),
-                                    room=sid,  # Emit to specific socket
-                                )
-                                emitted_count += 1
-                                logger.debug(f"Emitted {event_name} to socket {sid}")
-                        else:
-                            logger.warning(f"No user info found for socket {sid} in room {room_name}")
-                    except Exception as e:
-                        # Log and continue if we can't emit to this specific socket
-                        # (e.g., socket disconnected, session expired)
-                        logger.warning(f"Failed to emit event {event_name} to socket {sid}: {e}")
+                # Emit to the user's room
+                await self._sio.emit(
+                    event=event_name,
+                    data=event_data.model_dump(mode="json"),
+                    room=user_room
+                )
+                
+                # Also emit to admin room so admins can see all events
+                await self._sio.emit(
+                    event=event_name,
+                    data=event_data.model_dump(mode="json"),
+                    room="admin"
+                )
 
-                if emitted_count == 0:
-                    logger.warning(f"Event {event_name} was not emitted to any sockets (user_id: {event_data.user_id}, room: {room_name}, participants_checked: {participants_checked}, total_socket_users: {len(self._socket_users)})")
-                    logger.debug(f"Available socket users: {list(self._socket_users.keys())}")
+                logger.debug(f"Emitted {event_name} to user room {user_room} and admin room")
             else:
                 # For non-item events (like queue status), emit to all subscribers
                 await self._sio.emit(