Merge pull request #179 from wumibals/main

(ADMIN) — returns unassigned escalated queue

Author: BigBen-7

src/chat.py

@@ -63,6 +63,10 @@ def __init__(self) -> None:
         self.message_history: Dict[str, List[ChatMessage]] = {}
         self.escalations: List[EscalationEvent] = []
         self.user_connections: Dict[str, Set[str]] = {}
+        # status: "open" | "escalated" | "assigned" | "resolved"
+        self.conversation_statuses: Dict[str, str] = {}
+        self.conversation_assignments: Dict[str, Optional[str]] = {}
+        self.conversation_escalated_at: Dict[str, datetime] = {}
         self._lock = asyncio.Lock()
 
     async def connect(
@@ -192,6 +196,9 @@ async def escalate_conversation(
 
         async with self._lock:
             self.escalations.append(escalation)
+            self.conversation_statuses[conversation_id] = "escalated"
+            self.conversation_assignments[conversation_id] = None
+            self.conversation_escalated_at[conversation_id] = escalation.timestamp
 
         if conversation_id in self.active_connections:
             escalation_notification = {
@@ -225,6 +232,77 @@ async def escalate_conversation(
         )
         return escalation
 
+    async def assign_conversation(
+        self, conversation_id: str, agent_id: str
+    ) -> None:
+        """Assign an escalated conversation to a support agent."""
+        async with self._lock:
+            self.conversation_statuses[conversation_id] = "assigned"
+            self.conversation_assignments[conversation_id] = agent_id
+
+        notification = {
+            "type": "assignment",
+            "conversation_id": conversation_id,
+            "agent_id": agent_id,
+            "timestamp": datetime.utcnow().isoformat(),
+            "message": "This conversation has been assigned to a support agent.",
+        }
+
+        if conversation_id in self.active_connections:
+            disconnected: List[WebSocket] = []
+            for websocket in self.active_connections[conversation_id]:
+                try:
+                    await websocket.send_text(json.dumps(notification))
+                except Exception as exc:
+                    log_warning(
+                        "Failed to send assignment notification",
+                        {"conversation_id": conversation_id, "error": str(exc)},
+                    )
+                    disconnected.append(websocket)
+
+            if disconnected:
+                async with self._lock:
+                    for ws in disconnected:
+                        if ws in self.active_connections.get(conversation_id, []):
+                            self.active_connections[conversation_id].remove(ws)
+
+        log_info(
+            "Conversation assigned",
+            {"conversation_id": conversation_id, "agent_id": agent_id},
+        )
+
+    def get_conversation_status(
+        self, conversation_id: str
+    ) -> Dict[str, Any]:
+        """Return the current status and assignment for a conversation."""
+        status = self.conversation_statuses.get(conversation_id, "open")
+        assigned_agent_id = self.conversation_assignments.get(conversation_id)
+        return {
+            "conversation_id": conversation_id,
+            "status": status,
+            "assigned_agent_id": assigned_agent_id,
+        }
+
+    def get_unassigned_queue(self) -> List[Dict[str, Any]]:
+        """Return all escalated conversations with no assigned agent, ordered by escalated_at asc."""
+        queue = []
+        for conv_id, status in self.conversation_statuses.items():
+            if status == "escalated" and self.conversation_assignments.get(conv_id) is None:
+                escalated_at = self.conversation_escalated_at.get(conv_id)
+                # Find the most recent escalation reason for this conversation
+                reason = ""
+                for esc in reversed(self.escalations):
+                    if esc.conversation_id == conv_id:
+                        reason = esc.reason
+                        break
+                queue.append({
+                    "conversation_id": conv_id,
+                    "escalated_at": escalated_at,
+                    "reason": reason,
+                })
+        queue.sort(key=lambda x: x["escalated_at"] or datetime.min)
+        return queue
+
     def get_escalations(
         self, conversation_id: Optional[str] = None
     ) -> List[EscalationEvent]:

src/main.py

@@ -72,13 +72,18 @@
     AnalyticsTransfersResponse,
     HeatmapQuery,
     HeatmapResponse,
+    ChatAssignRequest,
+    ChatAssignResponse,
+    ChatConversationStatusResponse,
     ChatEscalateRequest,
     ChatEscalateResponse,
     ChatEscalationsResponse,
     ChatMessageHistoryQuery,
     ChatMessageHistoryResponse,
     ChatMessageSendRequest,
     ChatMessageSendResponse,
+    ChatQueueItem,
+    ChatQueueResponse,
     ChatTypingRequest,
     ChatTypingResponse,
     ChatUserConversationsResponse,
@@ -913,6 +918,75 @@ async def get_user_conversations(user_id: str) -> ChatUserConversationsResponse:
         raise HTTPException(status_code=500, detail="Failed to get user conversations")
 
 
+@app.get("/chat/queue", response_model=ChatQueueResponse)
+async def get_escalation_queue(
+    _: str = Depends(require_admin_key),
+) -> ChatQueueResponse:
+    """Return all unassigned escalated conversations ordered by escalation time (oldest first)."""
+    try:
+        raw_queue = chat_manager.get_unassigned_queue()
+        queue = [
+            ChatQueueItem(
+                conversation_id=item["conversation_id"],
+                escalated_at=item["escalated_at"],
+                reason=item["reason"],
+            )
+            for item in raw_queue
+        ]
+        return ChatQueueResponse(queue=queue, count=len(queue))
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("Error getting escalation queue: %s", exc)
+        raise HTTPException(status_code=500, detail="Failed to get escalation queue")
+
+
+@app.post("/chat/{conversation_id}/assign", response_model=ChatAssignResponse)
+async def assign_conversation(
+    conversation_id: str,
+    body: ChatAssignRequest,
+    _: str = Depends(require_admin_key),
+) -> ChatAssignResponse:
+    """Assign an escalated conversation to a support agent."""
+    try:
+        status_info = chat_manager.get_conversation_status(conversation_id)
+        if status_info["status"] not in ("escalated", "assigned"):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Conversation is not escalated (current status: {status_info['status']})",
+            )
+        await chat_manager.assign_conversation(conversation_id, body.agent_id)
+        return ChatAssignResponse(
+            status="success",
+            conversation_id=conversation_id,
+            agent_id=body.agent_id,
+        )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("Error assigning conversation: %s", exc)
+        raise HTTPException(status_code=500, detail="Failed to assign conversation")
+
+
+@app.get("/chat/{conversation_id}/status", response_model=ChatConversationStatusResponse)
+async def get_conversation_status(
+    conversation_id: str,
+) -> ChatConversationStatusResponse:
+    """Get the current status and agent assignment for a conversation."""
+    try:
+        info = chat_manager.get_conversation_status(conversation_id)
+        return ChatConversationStatusResponse(
+            conversation_id=info["conversation_id"],
+            status=info["status"],
+            assigned_agent_id=info["assigned_agent_id"],
+        )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("Error getting conversation status: %s", exc)
+        raise HTTPException(status_code=500, detail="Failed to get conversation status")
+
+
 # ---------------------------------------------------------------------------
 # Trending events
 # ---------------------------------------------------------------------------

src/types_custom.py

@@ -227,6 +227,38 @@ class ChatTypingResponse(BaseModel):
     status: Literal["success"]
 
 
+class ChatAssignRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    agent_id: str = Field(..., min_length=1)
+
+
+class ChatAssignResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    status: Literal["success"]
+    conversation_id: str
+    agent_id: str
+
+
+class ChatQueueItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    conversation_id: str
+    escalated_at: datetime
+    reason: str
+
+
+class ChatQueueResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    queue: List[ChatQueueItem]
+    count: int
+
+
+class ChatConversationStatusResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    conversation_id: str
+    status: Literal["open", "escalated", "assigned", "resolved"]
+    assigned_agent_id: Optional[str] = None
+
+
 class AnalyticsStatsQuery(BaseModel):
     model_config = ConfigDict(extra="forbid")
     event_id: Optional[str] = None