diff --git a/api/analytics.py b/api/analytics.py new file mode 100644 index 00000000..1dcdbed6 --- /dev/null +++ b/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) diff --git a/api/app_factory.py b/api/app_factory.py index 037a66bb..b338e197 100644 --- a/api/app_factory.py +++ b/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) diff --git a/api/core/text2sql.py b/api/core/text2sql.py index 57c18fb9..90cd9cdc 100644 --- a/api/core/text2sql.py +++ b/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 diff --git a/api/helpers/redaction.py b/api/helpers/redaction.py new file mode 100644 index 00000000..5e49e80c --- /dev/null +++ b/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["']?) + (?Ppassword|token|secret|api[_-]?key|authorization) + (?P=quote) + \s*[=:]\s* + (?P["']?) + (?:(?: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] diff --git a/api/routes/graphs.py b/api/routes/graphs.py index 7c1d1c54..38003faf 100644 --- a/api/routes/graphs.py +++ b/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,10 +265,16 @@ 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 @@ -257,7 +282,13 @@ async def stream(): 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", diff --git a/api/routes/usage_tracking.py b/api/routes/usage_tracking.py index 83ded999..8387a27a 100644 --- a/api/routes/usage_tracking.py +++ b/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: diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py new file mode 100644 index 00000000..5175a430 --- /dev/null +++ b/tests/test_error_analytics.py @@ -0,0 +1,90 @@ +"""Tests for QueryWeaver error analytics.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import asyncio +import pytest +from fastapi import Request + +from api import analytics + +pytestmark = [pytest.mark.unit] + + +def _request() -> Request: + request = Request({ + "type": "http", + "method": "POST", + "path": "/graphs/query", + "headers": [], + "query_string": b"", + "scheme": "https", + "server": ("queryweaver.example", 443), + }) + request.state.user_email = "user@example.com" + return request + + +def test_safe_message_redacts_credentials(): + """Persisted messages must not contain common credential values.""" + error = RuntimeError( + 'password=hunter2 token: abc123 Authorization: ****** ' + '"api_key": "json-secret" ******db.example:5432/app ' + '******host/db ' + 'redis://:redis-secret@cache.example:6379' + ) + + message = analytics._safe_message(error) # pylint: disable=protected-access + + for secret in ( + "hunter2", "abc123", "bearer-secret", "json-secret", + "pg-secret", "mysql-secret", + "redis-secret", + ): + assert secret not in message + + +def test_safe_message_preserves_diagnostic_phrases(): + """Words such as password and token are not secrets without a separator.""" + message = analytics._safe_message( # pylint: disable=protected-access + RuntimeError('password authentication failed; token expired') + ) + assert "password authentication failed" in message + assert "token expired" in message + + +@pytest.mark.asyncio +async def test_report_error_writes_org_graph(): + """Unhandled errors should be written with QueryWeaver attribution.""" + graph = AsyncMock() + client = MagicMock() + client.select_graph.return_value = graph + + sink: set = set() + with patch.object(analytics, "_DB_OVERRIDE", client): + analytics.report_error( + _request(), RuntimeError("sales demo failed"), task_sink=sink + ) + await asyncio.gather(*list(sink)) + + params = graph.query.await_args.args[1] + assert params == { + "type": "RuntimeError", + "message": "sales demo failed", + "endpoint": "/graphs/query", + "method": "POST", + "user_email": "user@example.com", + } + + +@pytest.mark.asyncio +async def test_report_error_swallows_setup_failure(): + """Analytics connection failures must preserve best-effort behavior.""" + sink: set = set() + broken_db = MagicMock() + broken_db.select_graph.side_effect = ConnectionError("offline") + with patch.object(analytics, "_DB_OVERRIDE", broken_db): + analytics.report_error(_request(), RuntimeError("boom"), task_sink=sink) + results = await asyncio.gather(*list(sink), return_exceptions=True) + + assert isinstance(results[0], ConnectionError) diff --git a/tests/test_usage_tracking.py b/tests/test_usage_tracking.py index 1d3a0973..c9f22f55 100644 --- a/tests/test_usage_tracking.py +++ b/tests/test_usage_tracking.py @@ -9,6 +9,7 @@ import asyncio import base64 import inspect +import uuid from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -64,7 +65,8 @@ async def test_records_successful_query_event(self): with patch.object(usage_tracking, "resolve_db", return_value=db), \ patch.object(usage_tracking, "is_general_graph", return_value=False): record_query_usage_background( - USER_ID, f"{USER_ID}_mydb", success=True, db=db, task_sink=sink + USER_ID, f"{USER_ID}_mydb", success=True, question="How many users?", + db=db, task_sink=sink ) await _drain(sink) @@ -73,11 +75,16 @@ async def test_records_successful_query_event(self): cypher, params = graph.query.await_args.args assert "MATCH (u:User {email: $email})" in cypher assert ":UsageEvent" in cypher + assert ":FAILED_WITH" in cypher + uuid.UUID(params.pop("query_id")) assert params == { "email": EMAIL, "graph_id": f"{USER_ID}_mydb", "is_demo": False, "success": True, + "question": "How many users?", + "error": "", + "endpoint": "", } @pytest.mark.asyncio @@ -87,12 +94,33 @@ async def test_records_failed_query_event(self): with patch.object(usage_tracking, "resolve_db", return_value=db), \ patch.object(usage_tracking, "is_general_graph", return_value=False): record_query_usage_background( - USER_ID, f"{USER_ID}_mydb", success=False, db=db, task_sink=sink + USER_ID, f"{USER_ID}_mydb", success=False, question="Broken query", + error="syntax error", db=db, task_sink=sink ) await _drain(sink) - _cypher, params = graph.query.await_args.args + cypher, params = graph.query.await_args.args + assert "CREATE (e)-[:FAILED_WITH]->(error)" in cypher + assert "CREATE (error:Error" in cypher assert params["success"] is False + assert params["question"] == "Broken query" + assert params["error"] == "syntax error" + assert "CASE WHEN $error = '' THEN 0 ELSE 1 END" in cypher + + @pytest.mark.asyncio + async def test_preserves_explicit_query_id(self): + """The route correlation ID must be persisted unchanged.""" + db, graph = _mock_db() + sink: set = set() + with patch.object(usage_tracking, "resolve_db", return_value=db), \ + patch.object(usage_tracking, "is_general_graph", return_value=False): + record_query_usage_background( + USER_ID, f"{USER_ID}_mydb", success=False, question="Broken", + error="failure", query_id="query-123", db=db, task_sink=sink + ) + await _drain(sink) + + assert graph.query.await_args.args[1]["query_id"] == "query-123" @pytest.mark.asyncio async def test_demo_graph_is_flagged(self): @@ -101,7 +129,8 @@ async def test_demo_graph_is_flagged(self): with patch.object(usage_tracking, "resolve_db", return_value=db), \ patch.object(usage_tracking, "is_general_graph", return_value=True): record_query_usage_background( - USER_ID, "DEMO_CRM", success=True, db=db, task_sink=sink + USER_ID, "DEMO_CRM", success=True, question="Demo question", + db=db, task_sink=sink ) await _drain(sink) @@ -115,7 +144,8 @@ async def test_invalid_user_id_skips_write(self): sink: set = set() with patch.object(usage_tracking, "resolve_db", return_value=db): record_query_usage_background( - "!!!bad!!!", "x_y", success=True, db=db, task_sink=sink + "!!!bad!!!", "x_y", success=True, question="Question", + db=db, task_sink=sink ) await _drain(sink) @@ -134,7 +164,8 @@ async def test_write_failure_is_swallowed(self): patch.object(usage_tracking.logging, "error") as mock_log_error: # The synchronous call must not raise despite the write failing. record_query_usage_background( - USER_ID, f"{USER_ID}_mydb", success=True, db=db, task_sink=sink + USER_ID, f"{USER_ID}_mydb", success=True, question="Question", + db=db, task_sink=sink ) await _drain(sink) @@ -150,6 +181,9 @@ def test_recorder_has_no_memory_or_provider_parameter(self): """Tracking cannot be gated by ``use_memory`` or the LLM provider: the recorder simply has no such inputs.""" params = set(inspect.signature(record_query_usage_background).parameters) - assert params == {"user_id", "namespaced", "success", "db", "task_sink"} + assert params == { + "user_id", "namespaced", "success", "question", "error", "query_id", + "endpoint", "db", "task_sink" + } assert "use_memory" not in params assert "provider" not in params