Collect QueryWeaver errors in GenAI analytics

api/analytics.py

@@ -0,0 +1,93 @@
+"""Best-effort error reporting to the organization analytics graph."""
+
+import asyncio
+import logging
+from typing import Optional
+
+from fastapi import Request
+
+from api.config import ORGANIZATIONS_GRAPH
+from api.helpers.redaction import redact_sensitive_text
+
+LOGGER = logging.getLogger(__name__)
+_DB_OVERRIDE = None
+
+
+def _safe_message(exc: Exception) -> str:
+    """Redact common credential forms before persisting an exception message."""
+    return redact_sensitive_text(str(exc))
+
+
+async def _write_error(request: Request, exc: Exception) -> None:
+    """Write one error to the organization graph."""
+    if _DB_OVERRIDE is None:
+        # pylint: disable=import-outside-toplevel
+        from api.core.db_resolver import resolve_db
+
+        database = resolve_db()
+    else:
+        database = _DB_OVERRIDE
+    graph = database.select_graph(ORGANIZATIONS_GRAPH)
+    user_email = getattr(request.state, "user_email", None)
+    await graph.query(
+        """
+        CREATE (e:Error {
+            source: 'queryweaver',
+            type: $type,
+            message: $message,
+            endpoint: $endpoint,
+            method: $method,
+            timestamp: timestamp()
+        })
+        WITH e
+        OPTIONAL MATCH (u:User {email: $user_email})
+        FOREACH (_ IN CASE WHEN u IS NULL THEN [] ELSE [1] END |
+            CREATE (u)-[:ENCOUNTERED]->(e)
+        )
+        """,
+        {
+            "type": type(exc).__name__,
+            "message": _safe_message(exc),
+            "endpoint": request.url.path,
+            "method": request.method,
+            "user_email": user_email,
+        },
+    )
+
+
+def report_error(
+    request: Request,
+    exc: Exception,
+    task_sink: Optional[set] = None,
+) -> None:
+    """Schedule an unhandled-error write without delaying the response."""
+    task = asyncio.create_task(_write_error(request, exc))
+    sink = task_sink
+    if sink is None:
+        try:
+            # pylint: disable=import-outside-toplevel
+            from api.core.pipeline import background_tasks_var
+
+            sink = background_tasks_var.get()
+        except ImportError:
+            sink = None
+    if sink is not None:
+        sink.add(task)
+        task.add_done_callback(sink.discard)
+
+    def _log_done(done: "asyncio.Task") -> None:
+        if done.cancelled():
+            return
+        analytics_error = done.exception()
+        if analytics_error is not None:
+            LOGGER.error(
+                "Failed to report QueryWeaver error to analytics: %s",
+                analytics_error,
+                exc_info=(
+                    type(analytics_error),
+                    analytics_error,
+                    analytics_error.__traceback__,
+                ),
+            )
+
+    task.add_done_callback(_log_done)

api/app_factory.py

@@ -17,6 +17,7 @@
 
 from api.auth.oauth_handlers import setup_oauth_handlers
 from api.auth.user_management import SECRET_KEY
+from api.analytics import report_error
 from api.routes.auth import auth_router, init_auth
 from api.routes.graphs import graphs_router
 from api.routes.database import database_router
@@ -329,6 +330,11 @@ async def handle_oauth_error(
         request: Request, exc: Exception
     ):  # pylint: disable=unused-argument
         """Handle OAuth-related errors gracefully"""
+        report_error(request, exc)
+        logging.exception(
+            "Unhandled error for %s %s", request.method, request.url.path
+        )
+
         # Check if it's an OAuth-related error
         # TODO check this scenario, pylint: disable=fixme
         if "token" in str(exc).lower() or "oauth" in str(exc).lower():
@@ -339,7 +345,7 @@ async def handle_oauth_error(
         if isinstance(exc, HTTPException):
             raise exc
 
-        # For other errors, let them bubble up
+        # Preserve the existing FastAPI 500 response behavior.
         raise exc
 
     # Serve React app for all non-API routes (SPA catch-all)

api/core/text2sql.py

@@ -82,7 +82,7 @@ class ConfirmRequest(BaseModel):
     """
     sql_query: str
     confirmation: str = ""
-    chat: list = []
+    chat: list[str] = []
     custom_api_key: str | None = None
     custom_model: str | None = None
     use_memory: bool = False

api/helpers/redaction.py

@@ -0,0 +1,30 @@
+"""Redact credentials from text before it is persisted or logged."""
+
+import re
+
+_SENSITIVE_VALUE = re.compile(
+    r"""(?ix)
+    (?P<quote>["']?)
+    (?P<key>password|token|secret|api[_-]?key|authorization)
+    (?P=quote)
+    \s*[=:]\s*
+    (?P<value_quote>["']?)
+    (?:(?:bearer|basic|token)\s+)?
+    [^\s,;}]+
+    (?P=value_quote)
+    """
+)
+_CONNECTION_PASSWORD = re.compile(
+    r"(?i)\b([a-z][a-z0-9+.\-]*://[^:@/\s]*:)[^@/\s]+@"
+)
+
+
+def redact_sensitive_text(value: str, limit: int = 4000) -> str:
+    """Remove common credential forms and embedded URL passwords."""
+    message = _SENSITIVE_VALUE.sub(
+        lambda match: f"{match.group('quote')}{match.group('key')}"
+        f"{match.group('quote')}: [REDACTED]",
+        value,
+    )
+    message = _CONNECTION_PASSWORD.sub(r"\1[REDACTED]@", message)
+    return message[:limit]

api/routes/graphs.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import uuid
 from fastapi import APIRouter, Request, HTTPException, UploadFile, File
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel
@@ -34,7 +35,9 @@
 graphs_router = APIRouter(tags=["Graphs & Databases"])
 
 
-async def _serialize_pipeline(gen, *, user_id, namespaced):
+async def _serialize_pipeline(  # pylint: disable=too-many-arguments
+    gen, *, user_id: str, namespaced: str, question: str, query_id: str, endpoint: str
+):
     """Serialize pipeline events to the wire format and stop on ``_Final``.
 
     Pure encoding loop — no exception handling here. Each route handler
@@ -61,6 +64,10 @@ async def _serialize_pipeline(gen, *, user_id, namespaced):
         record_query_usage_background(
             user_id, namespaced,
             success=final.is_valid and final.error_message is None,
+            question=question,
+            error=final.error_message or "",
+            query_id=query_id,
+            endpoint=endpoint,
         )
 
 
@@ -195,10 +202,16 @@ async def query_graph(
         return JSONResponse(content={"error": "Invalid query request"}, status_code=400)
 
     async def stream():
+        question = chat_data.chat[-1]
+        query_id = str(uuid.uuid4())
         try:
             async for chunk in _serialize_pipeline(
                 run_query(request.state.user_id, graph_id, chat_data),
-                user_id=request.state.user_id, namespaced=namespaced,
+                user_id=request.state.user_id,
+                namespaced=namespaced,
+                question=question,
+                query_id=query_id,
+                endpoint=request.url.path,
             ):
                 yield chunk
         except Exception:  # pylint: disable=broad-exception-caught
@@ -208,7 +221,13 @@ async def stream():
             # Pipeline crashed before _Final, so _serialize_pipeline didn't
             # record — count this attempt as a failure here.
             record_query_usage_background(
-                request.state.user_id, namespaced, success=False
+                request.state.user_id,
+                namespaced,
+                success=False,
+                question=question,
+                error="Unhandled streaming query failure",
+                query_id=query_id,
+                endpoint=request.url.path,
             )
             yield json.dumps({
                 "type": "error",
@@ -246,18 +265,30 @@ async def confirm_destructive_operation(
         return JSONResponse(content={"error": "Invalid confirmation request"}, status_code=400)
 
     async def stream():
+        question = str(confirm_data.chat[-1]) if confirm_data.chat else ""
+        query_id = str(uuid.uuid4())
         try:
             async for chunk in _serialize_pipeline(
                 run_confirmed(request.state.user_id, graph_id, confirm_data),
-                user_id=request.state.user_id, namespaced=namespaced,
+                user_id=request.state.user_id,
+                namespaced=namespaced,
+                question=question,
+                query_id=query_id,
+                endpoint=request.url.path,
             ):
                 yield chunk
         except Exception:  # pylint: disable=broad-exception-caught
             # See note on the query endpoint above (CodeQL).
             logging.exception("Streaming confirmed-destructive query failed")
             # Pipeline crashed before _Final — record the failed attempt here.
             record_query_usage_background(
-                request.state.user_id, namespaced, success=False
+                request.state.user_id,
+                namespaced,
+                success=False,
+                question=question,
+                error="Unhandled confirmed-query failure",
+                query_id=query_id,
+                endpoint=request.url.path,
             )
             yield json.dumps({
                 "type": "error",

api/routes/usage_tracking.py

@@ -13,8 +13,10 @@
   (``query_count``/``success_count``/``error_count``/``last_active``/
   ``first_query_at``) for cheap reads.
 * A per-query ``(:UsageEvent)`` node linked ``(User)-[:PERFORMED]->`` carrying
-  ``graph_id``/``is_demo``/``success``/``timestamp`` for time-series, per-DB
-  and success-rate analytics.
+  ``query_id``/``graph_id``/``is_demo``/``success``/``question``/``error``/
+  ``timestamp`` for time-series, per-DB, and success-rate analytics. Failed
+  executions also create ``(UsageEvent)-[:FAILED_WITH]->(Error)`` and
+  ``(User)-[:ENCOUNTERED]->(Error)`` relationships.
 
 Writes never block or fail a request: they run as background tasks whose
 exceptions are logged and swallowed, mirroring
@@ -26,11 +28,13 @@
 import binascii
 import hashlib
 import logging
+import uuid
 from typing import Optional
 
 from api.config import ORGANIZATIONS_GRAPH
 from api.core.db_resolver import resolve_db
 from api.core.pipeline import background_tasks_var, is_general_graph
+from api.helpers.redaction import redact_sensitive_text
 
 # Single round-trip: bump the User counters/timestamps and append a UsageEvent.
 # Uses MATCH (not MERGE) on User so an unknown email is a silent no-op rather
@@ -40,15 +44,30 @@
 MATCH (u:User {email: $email})
 SET u.query_count    = coalesce(u.query_count, 0) + 1,
     u.success_count  = coalesce(u.success_count, 0) + (CASE WHEN $success THEN 1 ELSE 0 END),
-    u.error_count    = coalesce(u.error_count, 0) + (CASE WHEN $success THEN 0 ELSE 1 END),
+    u.error_count    = coalesce(u.error_count, 0) + (CASE WHEN $error = '' THEN 0 ELSE 1 END),
     u.last_active    = timestamp(),
     u.first_query_at = coalesce(u.first_query_at, timestamp())
 CREATE (u)-[:PERFORMED]->(e:UsageEvent {
+    query_id: $query_id,
     graph_id: $graph_id,
     is_demo: $is_demo,
     success: $success,
+    question: $question,
+    error: $error,
     timestamp: timestamp()
 })
+FOREACH (_ IN CASE WHEN $error = '' THEN [] ELSE [1] END |
+    CREATE (error:Error {
+        source: 'queryweaver',
+        type: 'QueryError',
+        message: $error,
+        endpoint: $endpoint,
+        method: 'POST',
+        timestamp: timestamp()
+    })
+    CREATE (u)-[:ENCOUNTERED]->(error)
+    CREATE (e)-[:FAILED_WITH]->(error)
+)
 """
 
 
@@ -74,16 +93,30 @@ def _decode_email(user_id: str) -> Optional[str]:
     return email
 
 
-async def _write_usage(email: str, graph_id: str, is_demo: bool, success: bool, db) -> None:
+async def _write_usage(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    email: str,
+    query_id: str,
+    graph_id: str,
+    is_demo: bool,
+    success: bool,
+    question: str,
+    error: str,
+    endpoint: str,
+    db,
+) -> None:
     """Perform the single Cypher write against the Organizations graph."""
     organizations_graph = resolve_db(db).select_graph(ORGANIZATIONS_GRAPH)
     await organizations_graph.query(
         _RECORD_USAGE_CYPHER,
         {
             "email": email,
+            "query_id": query_id,
             "graph_id": graph_id,
             "is_demo": is_demo,
             "success": success,
+            "question": question,
+            "error": error,
+            "endpoint": endpoint,
         },
     )
     # Structured-ish log line so usage is visible to log aggregators even
@@ -98,10 +131,14 @@ async def _write_usage(email: str, graph_id: str, is_demo: bool, success: bool,
     )
 
 
-def record_query_usage_background(
+def record_query_usage_background(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     user_id: str,
     namespaced: str,
     success: bool,
+    question: str,
+    error: str = "",
+    query_id: Optional[str] = None,
+    endpoint: str = "",
     *,
     db=None,
     task_sink: Optional[set] = None,
@@ -118,6 +155,10 @@ def record_query_usage_background(
         namespaced: The fully-namespaced graph name the query ran against;
             already demo-aware, so it doubles as the recorded ``graph_id``.
         success: Whether SQL execution succeeded (no execution error).
+        question: The natural-language question associated with the attempt.
+        error: The pipeline or execution error for failed attempts.
+        query_id: Request-scoped identifier attached to the UsageEvent for correlation.
+        endpoint: Route path that handled the query.
         db: Optional FalkorDB handle; resolves to the server singleton when None.
         task_sink: Optional set the scheduled task is added to (and auto-removed
             from on completion) so callers can await any in-flight tracking
@@ -131,7 +172,17 @@ def record_query_usage_background(
     sink = task_sink if task_sink is not None else background_tasks_var.get()
 
     task = asyncio.create_task(
-        _write_usage(email, namespaced, is_demo, success, db)
+        _write_usage(
+            email,
+            query_id or str(uuid.uuid4()),
+            namespaced,
+            is_demo,
+            success,
+            redact_sensitive_text(str(question)),
+            redact_sensitive_text(error),
+            endpoint,
+            db,
+        )
     )
 
     if sink is not None: