Claude Code: Add CollaborationHub WebSocket message broker

- Add CollaborationHub class with room-based messaging (adam, eve, system, dialogue)
- Add /ws/hub WebSocket endpoint for room-based connections
- Add HTTP API endpoints: /api/hub/status, /api/hub/dialogue, /api/hub/broadcast
- Support inter-agent dialogue simulation and targeted room messaging

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

app.py

@@ -163,8 +163,254 @@ class ConnectionManager:
             self.disconnect(client_id)
 
 
-# Global connection manager instance
+# =============================================================================
+# CollaborationHub - Room-based WebSocket Message Broker
+# =============================================================================
+
+class CollaborationHub:
+    """
+    Centralized WebSocket message broker supporting distinct rooms/topics.
+
+    Features:
+    - Rooms: adam, eve, system, dialogue (for Agent Dialogue simulation)
+    - Automatic client connection tracking per room
+    - Broadcast to specific rooms or all rooms
+    - Targeted messaging to individual clients
+    """
+
+    # Available room types
+    ROOMS = {"adam", "eve", "system", "dialogue"}
+
+    def __init__(self) -> None:
+        # Room -> {client_id -> WebSocket}
+        self.rooms: Dict[str, Dict[str, WebSocket]] = {room: {} for room in self.ROOMS}
+        # Client tracking: client_id -> room
+        self.client_room: Dict[str, str] = {}
+        # Client info: client_id -> metadata
+        self.client_info: Dict[str, Dict[str, Any]] = {}
+
+    async def connect(self, client_id: str, websocket: WebSocket, room: str = "system") -> bool:
+        """
+        Connect a client to a specific room.
+
+        Args:
+            client_id: Unique identifier for the client
+            websocket: WebSocket connection object
+            room: Room to join (adam, eve, system, dialogue)
+
+        Returns:
+            True if connected, False if duplicate or invalid room
+        """
+        if room not in self.ROOMS:
+            logger.warning(
+                f"Invalid room: {room}",
+                extra={"client_id": client_id, "room": room, "action": "reject_invalid_room"}
+            )
+            return False
+
+        # Reject if client already connected to any room
+        if client_id in self.client_room:
+            existing_room = self.client_room[client_id]
+            logger.warning(
+                f"Client {client_id} already in room {existing_room}",
+                extra={"client_id": client_id, "existing_room": existing_room, "action": "reject_duplicate"}
+            )
+            return False
+
+        await websocket.accept()
+        self.rooms[room][client_id] = websocket
+        self.client_room[client_id] = room
+        self.client_info[client_id] = {
+            "room": room,
+            "connected_at": datetime.utcnow().isoformat()
+        }
+
+        logger.info(
+            f"Client {client_id} joined room: {room}",
+            extra={
+                "client_id": client_id,
+                "room": room,
+                "action": "join_room",
+                "room_members": len(self.rooms[room])
+            }
+        )
+
+        # Notify client of successful join
+        await websocket.send_json({
+            "type": "room_joined",
+            "room": room,
+            "client_id": client_id,
+            "timestamp": datetime.utcnow().isoformat()
+        })
+
+        return True
+
+    def disconnect(self, client_id: str) -> None:
+        """Disconnect a client from their room."""
+        if client_id not in self.client_room:
+            return
+
+        room = self.client_room[client_id]
+        if client_id in self.rooms[room]:
+            del self.rooms[room][client_id]
+
+        del self.client_room[client_id]
+        if client_id in self.client_info:
+            del self.client_info[client_id]
+
+        logger.info(
+            f"Client {client_id} left room: {room}",
+            extra={
+                "client_id": client_id,
+                "room": room,
+                "action": "leave_room",
+                "room_members": len(self.rooms[room])
+            }
+        )
+
+    async def send_to_room(self, room: str, message: Any) -> int:
+        """
+        Send a message to all clients in a specific room.
+
+        Args:
+            room: Target room name
+            message: Message payload (will be JSON serialized)
+
+        Returns:
+            Number of clients the message was sent to
+        """
+        if room not in self.ROOMS:
+            logger.warning(f"Cannot send to invalid room: {room}")
+            return 0
+
+        if not self.rooms[room]:
+            return 0
+
+        disconnected = []
+        sent_count = 0
+
+        for client_id, websocket in self.rooms[room].items():
+            try:
+                await websocket.send_json(message)
+                sent_count += 1
+            except Exception as e:
+                logger.warning(
+                    f"Failed to send to {client_id} in room {room}: {e}",
+                    extra={"client_id": client_id, "room": room, "error": str(e)}
+                )
+                disconnected.append(client_id)
+
+        # Clean up disconnected clients
+        for client_id in disconnected:
+            self.disconnect(client_id)
+
+        return sent_count
+
+    async def send_to_client(self, client_id: str, message: Any) -> bool:
+        """
+        Send a message to a specific client.
+
+        Args:
+            client_id: Target client ID
+            message: Message payload (will be JSON serialized)
+
+        Returns:
+            True if sent, False if client not found or send failed
+        """
+        if client_id not in self.client_room:
+            return False
+
+        room = self.client_room[client_id]
+        if client_id not in self.rooms[room]:
+            return False
+
+        try:
+            await self.rooms[room][client_id].send_json(message)
+            return True
+        except Exception as e:
+            logger.warning(
+                f"Failed to send to {client_id}: {e}",
+                extra={"client_id": client_id, "error": str(e)}
+            )
+            self.disconnect(client_id)
+            return False
+
+    async def broadcast(self, message: Any, exclude_room: str = None) -> int:
+        """
+        Broadcast a message to all rooms (or all except one).
+
+        Args:
+            message: Message payload (will be JSON serialized)
+            exclude_room: Optional room to exclude from broadcast
+
+        Returns:
+            Total number of clients the message was sent to
+        """
+        total_sent = 0
+        for room in self.ROOMS:
+            if exclude_room and room == exclude_room:
+                continue
+            total_sent += await self.send_to_room(room, message)
+        return total_sent
+
+    async def simulate_agent_dialogue(self, speaker: str, listener: str, message: str) -> None:
+        """
+        Simulate inter-agent dialogue by sending a message from one agent to another.
+
+        Args:
+            speaker: Room name of the speaking agent (adam or eve)
+            listener: Room name of the listening agent (adam or eve)
+            message: The message content
+        """
+        if speaker not in {"adam", "eve"} or listener not in {"adam", "eve"}:
+            logger.warning(f"Invalid agent dialogue: {speaker} -> {listener}")
+            return
+
+        dialogue_message = {
+            "type": "agent_dialogue",
+            "speaker": speaker,
+            "listener": listener,
+            "message": message,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+
+        # Send to listener's room
+        sent_count = await self.send_to_room(listener, dialogue_message)
+
+        # Also broadcast to system room for logging
+        await self.send_to_room("system", {
+            **dialogue_message,
+            "type": "agent_dialogue_log"
+        })
+
+        logger.info(
+            f"Agent dialogue: {speaker} -> {listener}: {message[:50]}...",
+            extra={
+                "speaker": speaker,
+                "listener": listener,
+                "recipients": sent_count,
+                "message_length": len(message)
+            }
+        )
+
+    def get_room_status(self) -> Dict[str, Any]:
+        """Return current status of all rooms."""
+        return {
+            room: {
+                "members": list(clients.keys()),
+                "count": len(clients)
+            }
+            for room, clients in self.rooms.items()
+        }
+
+    def get_client_info(self, client_id: str) -> Dict[str, Any]:
+        """Get info about a specific client."""
+        return self.client_info.get(client_id, {})
+
+
+# Global manager instances
 manager = ConnectionManager()
+hub = CollaborationHub()
 
 
 @app.exception_handler(Exception)
@@ -233,6 +479,107 @@ _worker_state = {
 }
 
 
+# =============================================================================
+# CollaborationHub HTTP API Endpoints
+# =============================================================================
+
+@app.get("/api/hub/status")
+@limiter.limit("60/minute")
+async def hub_status(request: Request):
+    """Get current CollaborationHub room status."""
+    return {
+        "timestamp": datetime.utcnow().isoformat(),
+        "rooms": hub.get_room_status(),
+        "total_clients": len(hub.client_room)
+    }
+
+
+@app.post("/api/hub/dialogue")
+@limiter.limit("30/minute")
+async def trigger_agent_dialogue(request: Request):
+    """
+    Trigger an agent dialogue simulation.
+
+    Body: {"speaker": "adam", "listener": "eve", "message": "Hello!"}
+    """
+    try:
+        payload = await request.json()
+        speaker = payload.get("speaker", "adam")
+        listener = payload.get("listener", "eve")
+        message = payload.get("message", "")
+
+        if not message:
+            return JSONResponse(
+                status_code=400,
+                content={"error": "message is required"}
+            )
+
+        await hub.simulate_agent_dialogue(speaker, listener, message)
+
+        return {
+            "status": "sent",
+            "speaker": speaker,
+            "listener": listener,
+            "message": message,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+
+    except Exception as e:
+        logger.error(f"Dialogue trigger error: {e}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={"error": str(e)}
+        )
+
+
+@app.post("/api/hub/broadcast")
+@limiter.limit("30/minute")
+async def broadcast_to_room(request: Request):
+    """
+    Broadcast a message to a specific room or all rooms.
+
+    Body: {"room": "adam", "data": {...}} or {"broadcast_all": true, "data": {...}}
+    """
+    try:
+        payload = await request.json()
+        data = payload.get("data", {})
+        room = payload.get("room")
+        broadcast_all = payload.get("broadcast_all", False)
+
+        if broadcast_all:
+            recipients = await hub.broadcast(data)
+            return {
+                "status": "broadcast",
+                "recipients": recipients,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        elif room:
+            if room not in hub.ROOMS:
+                return JSONResponse(
+                    status_code=400,
+                    content={"error": f"Invalid room. Available: {list(hub.ROOMS)}"}
+                )
+            recipients = await hub.send_to_room(room, data)
+            return {
+                "status": "sent",
+                "room": room,
+                "recipients": recipients,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        else:
+            return JSONResponse(
+                status_code=400,
+                content={"error": "Either 'room' or 'broadcast_all' is required"}
+            )
+
+    except Exception as e:
+        logger.error(f"Broadcast error: {e}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={"error": str(e)}
+        )
+
+
 @app.post("/internal/heartbeat")
 @limiter.limit("100/minute")  # Higher limit for internal heartbeat
 async def receive_heartbeat(request: Request):
@@ -369,6 +716,123 @@ async def status_check(request: Request):
     return {"status": status, "checks": checks}
 
 
+# =============================================================================
+# CollaborationHub WebSocket Endpoints
+# =============================================================================
+
+@app.websocket("/ws/hub")
+async def websocket_hub_endpoint(
+    websocket: WebSocket,
+    client_id: str = Query(..., description="Client identifier required"),
+    room: str = Query("system", description="Room to join: adam, eve, system, dialogue")
+) -> None:
+    """
+    CollaborationHub WebSocket endpoint with room-based messaging.
+
+    Rooms:
+    - adam: Adam agent messages and updates
+    - eve: Eve agent messages and updates
+    - system: System-wide notifications
+    - dialogue: Inter-agent dialogue simulation
+
+    Requires `client_id` and optionally `room` query parameters.
+    """
+    if not client_id:
+        await websocket.close(code=4008, reason="client_id query parameter required")
+        logger.warning("Hub connection rejected: missing client_id")
+        return
+
+    # Connect to the specified room
+    connected = await hub.connect(client_id, websocket, room)
+    if not connected:
+        await websocket.close(code=4009, reason=f"Failed to join room: {room}")
+        return
+
+    try:
+        while True:
+            # Receive and handle JSON message from client
+            try:
+                data = await websocket.receive_json()
+                msg_type = data.get("type", "unknown")
+
+                logger.info(
+                    f"Hub message from {client_id} in room {room}: {msg_type}",
+                    extra={"client_id": client_id, "room": room, "msg_type": msg_type}
+                )
+
+                # Handle different message types
+                if msg_type == "broadcast":
+                    # Broadcast to all rooms
+                    response = {
+                        "type": "broadcast_result",
+                        "from_client": client_id,
+                        "from_room": room,
+                        "data": data.get("data"),
+                        "recipients": await hub.broadcast(data, exclude_room=None),
+                        "timestamp": datetime.utcnow().isoformat()
+                    }
+                    await websocket.send_json(response)
+
+                elif msg_type == "room_message":
+                    # Send to specific room
+                    target_room = data.get("target_room", "system")
+                    recipients = await hub.send_to_room(target_room, data.get("data", {}))
+                    await websocket.send_json({
+                        "type": "room_message_result",
+                        "target_room": target_room,
+                        "recipients": recipients,
+                        "timestamp": datetime.utcnow().isoformat()
+                    })
+
+                elif msg_type == "agent_dialogue":
+                    # Simulate agent-to-agent dialogue
+                    speaker = data.get("speaker", room)
+                    listener = data.get("listener", "eve" if speaker == "adam" else "adam")
+                    message = data.get("message", "")
+                    await hub.simulate_agent_dialogue(speaker, listener, message)
+
+                elif msg_type == "status":
+                    # Get room status
+                    await websocket.send_json({
+                        "type": "room_status",
+                        "status": hub.get_room_status(),
+                        "your_room": room,
+                        "your_info": hub.get_client_info(client_id),
+                        "timestamp": datetime.utcnow().isoformat()
+                    })
+
+                else:
+                    # Echo back acknowledgment
+                    await websocket.send_json({
+                        "type": "ack",
+                        "client_id": client_id,
+                        "room": room,
+                        "received_type": msg_type,
+                        "timestamp": datetime.utcnow().isoformat()
+                    })
+
+            except json.JSONDecodeError as e:
+                logger.warning(
+                    f"Invalid JSON from {client_id}: {e}",
+                    extra={"client_id": client_id, "error": str(e)}
+                )
+                await websocket.send_json({"error": "invalid_json", "message": "Failed to parse JSON message"})
+
+    except WebSocketDisconnect:
+        logger.info(f"Hub WebSocket disconnected: {client_id}", extra={"client_id": client_id, "room": room})
+
+    except Exception as e:
+        logger.error(
+            f"Hub WebSocket error for {client_id}: {e}",
+            extra={"client_id": client_id, "room": room, "error": str(e)},
+            exc_info=True
+        )
+
+    finally:
+        # Always clean up connection
+        hub.disconnect(client_id)
+
+
 @app.websocket("/ws")
 async def websocket_endpoint(
     websocket: WebSocket,