Improve backend error handling and logging for async tasks

backend/app/agents/base_agent.py

@@ -43,12 +43,12 @@ async def run(self, initial_state: AgentState, thread_id: str) -> AgentState:
             result = await self.graph.ainvoke(initial_state.model_dump(), config)
             return AgentState(**result)
         except AttributeError as e:
-            logger.error("Graph not properly initialized for %s: %s", self.agent_name, str(e))
+            logger.error("Graph not properly initialized for %s: %s", self.agent_name, str(e), exc_info=True)
             state_dict = initial_state.model_dump()
             state_dict['errors'].append(f"Agent initialization error: {str(e)}")
             return AgentState(**state_dict)
         except Exception as e:
-            logger.error("Error in %s: %s", self.agent_name, str(e))
+            logger.error("Error in %s: %s", self.agent_name, str(e), exc_info=True)
             state_dict = initial_state.model_dump()
             state_dict['errors'].append(str(e))
             return AgentState(**state_dict)
@@ -74,5 +74,5 @@ async def stream_run(self, initial_state: AgentState, thread_id: str) -> AsyncGe
 
             logger.info(f"Streaming completed after {step_count} steps")
         except Exception as e:
-            logger.error("Error in %s stream: %s", self.agent_name, str(e))
+            logger.error("Error in %s stream: %s", self.agent_name, str(e), exc_info=True)
             yield {"error": str(e)}

backend/app/api/v1/integrations.py

@@ -9,8 +9,10 @@
 )
 from app.services.integration_service import integration_service, IntegrationNotFoundError
 from app.core.dependencies import get_current_user
+import logging
 
 router = APIRouter()
+logger = logging.getLogger(__name__)
 
 
 @router.post("/", response_model=IntegrationResponse, status_code=status.HTTP_201_CREATED)
@@ -24,7 +26,8 @@ async def create_integration(
     except ValueError as e:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
     except Exception as e:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+        logger.error("Failed to create integration for user %s", user_id, exc_info=True)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create integration") from e
 
 
 @router.get("/", response_model=IntegrationListResponse)
@@ -34,7 +37,8 @@ async def list_integrations(user_id: UUID = Depends(get_current_user)):
         integrations = await integration_service.get_integrations(user_id)
         return IntegrationListResponse(integrations=integrations, total=len(integrations))
     except Exception as e:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+        logger.error("Failed to list integrations for user %s", user_id, exc_info=True)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to list integrations") from e
 
 
 @router.get("/status/{platform}", response_model=IntegrationStatusResponse)
@@ -46,7 +50,8 @@ async def get_integration_status(
     try:
         return await integration_service.get_integration_status(user_id, platform)
     except Exception as e:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+        logger.error("Failed to get integration status for user %s platform %s", user_id, platform, exc_info=True)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to get integration status") from e
 
 @router.get("/{integration_id}", response_model=IntegrationResponse)
 async def get_integration(
@@ -64,8 +69,11 @@ async def get_integration(
             )
 
         return integration
+    except HTTPException:
+        raise
     except Exception as e:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+        logger.error("Failed to get integration %s for user %s", integration_id, user_id, exc_info=True)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to get integration") from e
 
 @router.put("/{integration_id}", response_model=IntegrationResponse)
 async def update_integration(
@@ -85,7 +93,7 @@ async def update_integration(
     except Exception as e:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to update integration: {str(e)}"
+            detail="Failed to update integration"
         ) from e
 
 @router.delete("/{integration_id}", status_code=status.HTTP_204_NO_CONTENT)
@@ -105,5 +113,5 @@ async def delete_integration(
     except Exception as e:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to delete integration: {str(e)}"
+            detail="Failed to delete integration"
         ) from e

backend/app/core/config/settings.py

@@ -19,6 +19,7 @@ class Settings(BaseSettings):
     # DB configuration
     supabase_url: str
     supabase_key: str
+    database_url: Optional[str] = None
 
     # LangSmith Tracing
     langsmith_tracing: bool = False

backend/app/core/events/event_bus.py

@@ -14,6 +14,7 @@ def __init__(self, handler_registry: HandlerRegistry):
         self.handler_registry = handler_registry
         self.handlers: Dict[EventType, List[callable]] = {}
         self.global_handlers: List[callable] = []
+        self._background_tasks: set[asyncio.Task] = set()
 
     def register_handler(self, event_type: Union[EventType, List[EventType]], handler_func, platform: Optional[PlatformType] = None):
         """Register a handler function for a specific event type and optionally platform"""
@@ -39,16 +40,31 @@ def register_global_handler(self, handler_func):
     async def dispatch(self, event: BaseEvent):
         """Dispatch an event to all registered handlers"""
 
+        async def _safe_call(handler, event):
+            """Wrapper to catch and log exceptions from fire-and-forget handler tasks."""
+            try:
+                await handler(event)
+            except Exception:
+                logger.error(
+                    "Event handler '%s' failed for event %s (type: %s)",
+                    handler.__name__, event.id, event.event_type,
+                    exc_info=True,
+                )
+
         # Call global handlers first
         for handler in self.global_handlers:
             logger.info(f"Calling global handler: {handler.__name__}")
-            asyncio.create_task(handler(event))
+            task = asyncio.create_task(_safe_call(handler, event))
+            self._background_tasks.add(task)
+            task.add_done_callback(self._background_tasks.discard)
 
         # Call event-specific handlers
         if event.event_type in self.handlers:
             for handler in self.handlers[event.event_type]:
                 logger.info(f"Calling handler: {handler.__name__} for event type: {event.event_type}")
-                asyncio.create_task(handler(event))
+                task = asyncio.create_task(_safe_call(handler, event))
+                self._background_tasks.add(task)
+                task.add_done_callback(self._background_tasks.discard)
         else:
             logger.info(f"No handlers registered for event type {event.event_type}")
             pass

backend/app/core/handler/faq_handler.py

@@ -43,7 +43,9 @@ async def _handle_faq_request(self, event: BaseEvent) -> Dict[str, Any]:
 
         logger.info("Question: " + question + ", Response: " + response)
 
-        await self._send_discord_response(event.channel_id, response)
+        sent = await self._send_discord_response(event.channel_id, response)
+        if not sent:
+            return {"success": False, "reason": "Failed to send FAQ response"}
         return {"success": True, "action": "faq_response_sent"}
 
     def get_faq_response(self, question: str) -> str:
@@ -53,11 +55,21 @@ async def _handle_knowledge_update(self, event: BaseEvent) -> Dict[str, Any]:
         """Handles knowledge base updates."""
         return {"success": True, "action": "knowledge_updated"}
 
-    async def _send_discord_response(self, channel_id: str, response: str):
+    async def _send_discord_response(self, channel_id: str, response: str) -> bool:
         """Sends a response message to the specified Discord channel."""
         if self.bot:
-            channel = self.bot.get_channel(int(channel_id))
-            if channel:
-                await channel.send(response)
-            else:
-                logger.error(f"Could not find Discord channel with ID {channel_id}")
+            try:
+                channel = self.bot.get_channel(int(channel_id))
+                if channel:
+                    await channel.send(response)
+                    return True
+                else:
+                    logger.error("Could not find Discord channel with ID %s", channel_id)
+                    return False
+            except Exception:
+                logger.error(
+                    "Failed to send Discord response to channel %s",
+                    channel_id, exc_info=True,
+                )
+                return False
+        return False

backend/app/core/handler/handler_registry.py

@@ -19,13 +19,17 @@ def register(self, event_types: List[EventType], handler_class: Type[BaseHandler
 
     def get_handler(self, event: BaseEvent) -> BaseHandler:
         """Get handler instance for an event"""
+        # Handle both enum and string values for platform and event_type
+        platform_val = event.platform.value if hasattr(event.platform, 'value') else event.platform
+        event_type_val = event.event_type.value if hasattr(event.event_type, 'value') else event.event_type
+
         # Try platform-specific handler first
-        key = f"{event.platform.value}:{event.event_type.value}"
+        key = f"{platform_val}:{event_type_val}"
         handler_class = self.handlers.get(key)
 
         # Fall back to generic event type handler
         if not handler_class:
-            key = event.event_type.value
+            key = event_type_val
             handler_class = self.handlers.get(key)
 
         if not handler_class:

backend/app/core/handler/message_handler.py

@@ -43,10 +43,6 @@ async def _handle_message_created(self, event: BaseEvent) -> Dict[str, Any]:
             )
             return await self.faq_handler.handle(faq_event)
 
-        # Implementation for new message creation
-        # - Check if it's a command
-        # - Check if it's a question
-        # - Process natural language
         return {"success": True, "action": "message_processed"}
 
     async def _handle_message_updated(self, event: BaseEvent) -> Dict[str, Any]:

backend/app/core/orchestration/agent_coordinator.py

@@ -27,6 +27,7 @@ def _register_handlers(self):
     @traceable(name="devrel_request_coordination", run_type="chain")
     async def _handle_devrel_request(self, message_data: Dict[str, Any]):
         """Handle DevRel agent requests"""
+        session_id = None
         try:
             # Extract memory thread ID (user_id for Discord)
             memory_thread_id = message_data.get("memory_thread_id") or message_data.get("user_id", "")
@@ -58,11 +59,15 @@ async def _handle_devrel_request(self, message_data: Dict[str, Any]):
                 await self._send_response_to_platform(message_data, result_state.final_response)
 
         except Exception as e:
-            logger.error(f"Error handling DevRel request: {str(e)}")
+            logger.error(
+                "Error handling DevRel request for session %s",
+                session_id, exc_info=True,
+            )
             await self._send_error_response(message_data, "I'm having trouble processing your request. Please try again.")
 
     async def _handle_clear_memory_request(self, message_data: Dict[str, Any]):
         """Handle requests to clear thread memory"""
+        memory_thread_id = None
         try:
             memory_thread_id = message_data.get("memory_thread_id")
             cleanup_reason = message_data.get("cleanup_reason", "manual")
@@ -81,8 +86,11 @@ async def _handle_clear_memory_request(self, message_data: Dict[str, Any]):
             else:
                 logger.error(f"Failed to clear memory for thread {memory_thread_id}")
 
-        except Exception as e:
-            logger.error(f"Error clearing memory: {str(e)}")
+        except Exception:
+            logger.error(
+                "Error clearing memory for thread %s",
+                memory_thread_id, exc_info=True,
+            )
 
     async def _handle_memory_timeout(self, memory_thread_id: str, state: AgentState):
         """Handle memory timeout - store to database and clear from InMemorySaver"""
@@ -97,11 +105,15 @@ async def _handle_memory_timeout(self, memory_thread_id: str, state: AgentState)
 
             logger.info(f"Memory timeout handled successfully for thread {memory_thread_id}")
 
-        except Exception as e:
-            logger.error(f"Error handling memory timeout: {str(e)}")
+        except Exception:
+            logger.error(
+                "Error handling memory timeout for thread %s",
+                memory_thread_id, exc_info=True,
+            )
 
     async def _send_response_to_platform(self, original_message: Dict[str, Any], response: str):
         """Send agent response back to the originating platform"""
+        platform = None
         try:
             platform = original_message.get("platform", "discord")
 
@@ -116,8 +128,11 @@ async def _send_response_to_platform(self, original_message: Dict[str, Any], res
 
                 await self.queue_manager.enqueue(response_message)
 
-        except Exception as e:
-            logger.error(f"Error sending response to platform: {str(e)}")
+        except Exception:
+            logger.error(
+                "Error sending response to platform %s",
+                platform, exc_info=True,
+            )
 
     async def _send_error_response(self, original_message: Dict[str, Any], error_message: str):
         """Send error response to platform"""

backend/app/core/orchestration/queue_manager.py

@@ -40,8 +40,8 @@ async def connect(self):
             for queue_name in self.queues.values():
                 await self.channel.declare_queue(queue_name, durable=True)
             logger.info("Successfully connected to RabbitMQ")
-        except Exception as e:
-            logger.error(f"Failed to connect to RabbitMQ: {e}")
+        except Exception:
+            logger.error("Failed to connect to RabbitMQ", exc_info=True)
             raise
 
     async def start(self, num_workers: int = 3):
@@ -75,6 +75,8 @@ async def enqueue(self,
                       priority: QueuePriority = QueuePriority.MEDIUM,
                       delay: float = 0):
         """Add a message to the queue"""
+        if self.channel is None:
+            raise RuntimeError("Cannot enqueue: RabbitMQ channel is not connected. Call connect() first.")
 
         if delay > 0:
             await asyncio.sleep(delay)
@@ -84,12 +86,19 @@ async def enqueue(self,
             "priority": priority,
             "data": message
         }
-        json_message = json.dumps(queue_item).encode()
-        await self.channel.default_exchange.publish(
-            aio_pika.Message(body=json_message),
-            routing_key=self.queues[priority]
-        )
-        logger.info(f"Enqueued message {queue_item['id']} with priority {priority}")
+        try:
+            json_message = json.dumps(queue_item).encode()
+            await self.channel.default_exchange.publish(
+                aio_pika.Message(body=json_message),
+                routing_key=self.queues[priority]
+            )
+            logger.info(f"Enqueued message {queue_item['id']} with priority {priority}")
+        except Exception:
+            logger.error(
+                "Failed to enqueue message %s with priority %s",
+                queue_item['id'], priority, exc_info=True,
+            )
+            raise
 
     def register_handler(self, message_type: str, handler: Callable):
         """Register a handler for a specific message type"""
@@ -113,18 +122,22 @@ async def _worker(self, worker_name: str):
                             item = json.loads(message.body.decode())
                             await self._process_item(item, worker_name)
                             await message.ack()
-                        except Exception as e:
-                            logger.error(f"Error processing message: {e}")
+                        except Exception:
+                            logger.error(
+                                "Error processing message in worker %s",
+                                worker_name, exc_info=True,
+                            )
                             await message.nack(requeue=False)
                 except asyncio.CancelledError:
                     logger.info(f"Worker {worker_name} cancelled")
                     return
-                except Exception as e:
-                    logger.error(f"Worker {worker_name} error: {e}")
+                except Exception:
+                    logger.error("Worker %s error", worker_name, exc_info=True)
             await asyncio.sleep(0.1)
 
     async def _process_item(self, item: Dict[str, Any], worker_name: str):
         """Process a queue item"""
+        message_type = "unknown"
         try:
             message_data = item["data"]
             message_type = message_data.get("type", "unknown")
@@ -140,5 +153,10 @@ async def _process_item(self, item: Dict[str, Any], worker_name: str):
             else:
                 logger.warning(f"No handler found for message type: {message_type}")
 
-        except Exception as e:
-            logger.error(f"Error processing item {item.get('id', 'unknown')}: {str(e)}")
+        except Exception:
+            logger.error(
+                "Error processing item %s (type: %s) in worker %s",
+                item.get('id', 'unknown'), message_type, worker_name,
+                exc_info=True,
+            )
+            raise