Spaces:
Running
Running
| # app/core/error_store.py | |
| """ | |
| Fire-and-forget error log recorder. | |
| Call record_error() anywhere an exception is caught and you want full | |
| context stored in MongoDB for the admin dashboard error-log view. | |
| Schema per document: | |
| error_type str exception class name | |
| error_msg str str(exc) | |
| stack_trace str full traceback | |
| agent str | None | |
| tool str | None | |
| feature str | None | |
| user_id str | None | |
| session_id str | None | |
| input_preview str | None first 500 chars of the triggering input | |
| resolved bool default False — admin can mark True | |
| ts datetime (UTC) | |
| """ | |
| import traceback | |
| from datetime import datetime, timezone | |
| from typing import Optional | |
| import structlog | |
| logger = structlog.get_logger(__name__) | |
| async def record_error( | |
| exc: Exception, | |
| *, | |
| agent: Optional[str] = None, | |
| tool: Optional[str] = None, | |
| feature: Optional[str] = None, | |
| user_id: Optional[str] = None, | |
| session_id: Optional[str] = None, | |
| input_preview: Optional[str] = None, | |
| ) -> None: | |
| """ | |
| Write one error event to MongoDB. Non-critical — never raises. | |
| Call with asyncio.create_task() so it never blocks the response path. | |
| """ | |
| try: | |
| from app.database import get_db | |
| db = await get_db() | |
| tb = traceback.format_exc() or "" | |
| await db["error_logs"].insert_one({ | |
| "error_type": type(exc).__name__, | |
| "error_msg": str(exc)[:1000], | |
| "stack_trace": tb[:5000], | |
| "agent": agent, | |
| "tool": tool, | |
| "feature": feature, | |
| "user_id": user_id, | |
| "session_id": session_id, | |
| "input_preview": (input_preview or "")[:500] if input_preview else None, | |
| "resolved": False, | |
| "ts": datetime.now(timezone.utc), | |
| }) | |
| except Exception as inner: | |
| logger.warning("error_store: insert failed", error=str(inner)[:200]) | |