From 302ea656f6554093a0d4f8c7fb3802ec9b7783a3 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:25:23 +0300 Subject: [PATCH 1/8] Collect unhandled errors in org analytics Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 55 +++++++++++++++++++++++++++++++++++ api/app_factory.py | 8 ++++- tests/test_error_analytics.py | 47 ++++++++++++++++++++++++++++++ 3 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 api/analytics.py create mode 100644 tests/test_error_analytics.py diff --git a/api/analytics.py b/api/analytics.py new file mode 100644 index 00000000..9168f554 --- /dev/null +++ b/api/analytics.py @@ -0,0 +1,55 @@ +"""Best-effort error reporting to the organization analytics graph.""" + +import logging +import os + +from fastapi import Request +from falkordb.asyncio import FalkorDB +from redis.asyncio import BlockingConnectionPool + +LOGGER = logging.getLogger(__name__) +ANALYTICS_GRAPH = os.getenv("ORGANIZATIONS_GRAPH", "Organizations") + + +async def report_error(request: Request, exc: Exception) -> bool: + """Record an unhandled QueryWeaver error without affecting the response.""" + url = os.getenv("FALKORDB_URL") + if not url: + LOGGER.error("Cannot report QueryWeaver error: FALKORDB_URL is not configured") + return False + + pool = BlockingConnectionPool.from_url(url, decode_responses=True) + client = FalkorDB(connection_pool=pool) + try: + graph = client.select_graph(ANALYTICS_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": str(exc)[:4000], + "endpoint": request.url.path, + "method": request.method, + "user_email": user_email, + }, + ) + return True + except Exception as analytics_error: # pylint: disable=broad-exception-caught + LOGGER.error("Failed to report QueryWeaver error to analytics: %s", analytics_error) + return False + finally: + await pool.aclose() diff --git a/api/app_factory.py b/api/app_factory.py index 037a66bb..060b4449 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 @@ -339,7 +340,12 @@ async def handle_oauth_error( if isinstance(exc, HTTPException): raise exc - # For other errors, let them bubble up + await report_error(request, exc) + logging.exception( + "Unhandled error for %s %s", request.method, request.url.path, exc_info=exc + ) + + # Preserve the existing FastAPI 500 response behavior. raise exc # Serve React app for all non-API routes (SPA catch-all) diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py new file mode 100644 index 00000000..1633735e --- /dev/null +++ b/tests/test_error_analytics.py @@ -0,0 +1,47 @@ +"""Tests for QueryWeaver error analytics.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import Request + +from api import analytics + + +@pytest.mark.asyncio +async def test_report_error_writes_org_graph(monkeypatch): + """Unhandled errors should be written with QueryWeaver attribution.""" + monkeypatch.setenv("FALKORDB_URL", "redis://analytics.example:6379") + graph = AsyncMock() + client = MagicMock() + client.select_graph.return_value = graph + pool = AsyncMock() + scope = { + "type": "http", + "method": "POST", + "path": "/graphs/query", + "headers": [], + "query_string": b"", + "scheme": "https", + "server": ("queryweaver.example", 443), + } + request = Request(scope) + request.state.user_email = "user@example.com" + + with ( + patch.object(analytics.BlockingConnectionPool, "from_url", return_value=pool), + patch.object(analytics, "FalkorDB", return_value=client), + ): + result = await analytics.report_error(request, RuntimeError("sales demo failed")) + + assert result is True + graph.query.assert_awaited_once() + 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", + } + pool.aclose.assert_awaited_once() From 3dfda2c03f53291c90c309ece6e97b29522608b8 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:25:23 +0300 Subject: [PATCH 2/8] Link failed queries to analytics errors Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/routes/graphs.py | 34 +++++++++++++++++++++---- api/routes/usage_tracking.py | 49 +++++++++++++++++++++++++++++++++--- 2 files changed, 75 insertions(+), 8 deletions(-) diff --git a/api/routes/graphs.py b/api/routes/graphs.py index 7c1d1c54..fdf9700f 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,7 @@ graphs_router = APIRouter(tags=["Graphs & Databases"]) -async def _serialize_pipeline(gen, *, user_id, namespaced): +async def _serialize_pipeline(gen, *, user_id, namespaced, question, query_id): """Serialize pipeline events to the wire format and stop on ``_Final``. Pure encoding loop — no exception handling here. Each route handler @@ -61,6 +62,9 @@ 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, ) @@ -195,10 +199,15 @@ 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, ): yield chunk except Exception: # pylint: disable=broad-exception-caught @@ -208,7 +217,12 @@ 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, ) yield json.dumps({ "type": "error", @@ -246,10 +260,15 @@ async def confirm_destructive_operation( return JSONResponse(content={"error": "Invalid confirmation request"}, status_code=400) async def stream(): + question = 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, ): yield chunk except Exception: # pylint: disable=broad-exception-caught @@ -257,7 +276,12 @@ 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, ) yield json.dumps({ "type": "error", diff --git a/api/routes/usage_tracking.py b/api/routes/usage_tracking.py index 83ded999..5966bfe1 100644 --- a/api/routes/usage_tracking.py +++ b/api/routes/usage_tracking.py @@ -26,6 +26,7 @@ import binascii import hashlib import logging +import uuid from typing import Optional from api.config import ORGANIZATIONS_GRAPH @@ -44,11 +45,26 @@ 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 $success THEN [] ELSE [1] END | + CREATE (error:Error { + source: 'queryweaver', + type: 'QueryError', + message: CASE WHEN $error = '' THEN 'Query could not be completed' ELSE $error END, + endpoint: '/graphs/' + $graph_id, + method: 'POST', + timestamp: timestamp() + }) + CREATE (u)-[:ENCOUNTERED]->(error) + CREATE (e)-[:FAILED_WITH]->(error) +) """ @@ -74,16 +90,28 @@ 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, + 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, }, ) # Structured-ish log line so usage is visible to log aggregators even @@ -98,10 +126,13 @@ 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, *, db=None, task_sink: Optional[set] = None, @@ -118,6 +149,9 @@ 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 used to link an unhandled error. 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 +165,16 @@ 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, + question[:4000], + error[:4000], + db, + ) ) if sink is not None: From dc73d137ec5838117ab82c67daa5d5270fdba1ae Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:25:23 +0300 Subject: [PATCH 3/8] Test query error analytics tracking Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_usage_tracking.py | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/tests/test_usage_tracking.py b/tests/test_usage_tracking.py index 1d3a0973..6de84158 100644 --- a/tests/test_usage_tracking.py +++ b/tests/test_usage_tracking.py @@ -64,7 +64,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 +74,15 @@ 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 assert params == { "email": EMAIL, + "query_id": params["query_id"], "graph_id": f"{USER_ID}_mydb", "is_demo": False, "success": True, + "question": "How many users?", + "error": "", } @pytest.mark.asyncio @@ -87,12 +92,15 @@ 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 assert params["success"] is False + assert params["question"] == "Broken query" + assert params["error"] == "syntax error" @pytest.mark.asyncio async def test_demo_graph_is_flagged(self): @@ -101,7 +109,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 +124,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 +144,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 +161,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", + "db", "task_sink" + } assert "use_memory" not in params assert "provider" not in params From 03b9f0b0329739af9b3d602534ac91f45fd5c4bd Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:31:25 +0300 Subject: [PATCH 4/8] Keep analytics error reporting best effort Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 11 ++++++++--- tests/test_error_analytics.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/api/analytics.py b/api/analytics.py index 9168f554..3cbf53d9 100644 --- a/api/analytics.py +++ b/api/analytics.py @@ -18,9 +18,10 @@ async def report_error(request: Request, exc: Exception) -> bool: LOGGER.error("Cannot report QueryWeaver error: FALKORDB_URL is not configured") return False - pool = BlockingConnectionPool.from_url(url, decode_responses=True) - client = FalkorDB(connection_pool=pool) + pool = None try: + pool = BlockingConnectionPool.from_url(url, decode_responses=True) + client = FalkorDB(connection_pool=pool) graph = client.select_graph(ANALYTICS_GRAPH) user_email = getattr(request.state, "user_email", None) await graph.query( @@ -52,4 +53,8 @@ async def report_error(request: Request, exc: Exception) -> bool: LOGGER.error("Failed to report QueryWeaver error to analytics: %s", analytics_error) return False finally: - await pool.aclose() + if pool is not None: + try: + await pool.aclose() + except Exception: # pylint: disable=broad-exception-caught + LOGGER.debug("Failed to close analytics Redis pool", exc_info=True) diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py index 1633735e..6b526671 100644 --- a/tests/test_error_analytics.py +++ b/tests/test_error_analytics.py @@ -7,6 +7,8 @@ from api import analytics +pytestmark = [pytest.mark.unit] + @pytest.mark.asyncio async def test_report_error_writes_org_graph(monkeypatch): @@ -45,3 +47,32 @@ async def test_report_error_writes_org_graph(monkeypatch): "user_email": "user@example.com", } pool.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_report_error_without_url_returns_false(monkeypatch): + """Missing analytics configuration must not mask the original error.""" + monkeypatch.delenv("FALKORDB_URL", raising=False) + request = Request({ + "type": "http", "method": "GET", "path": "/", "headers": [], + "query_string": b"", "scheme": "http", "server": ("localhost", 80), + }) + + assert await analytics.report_error(request, RuntimeError("boom")) is False + + +@pytest.mark.asyncio +async def test_report_error_swallows_setup_failure(monkeypatch): + """Invalid connection configuration must preserve best-effort behavior.""" + monkeypatch.setenv("FALKORDB_URL", "invalid") + request = Request({ + "type": "http", "method": "GET", "path": "/", "headers": [], + "query_string": b"", "scheme": "http", "server": ("localhost", 80), + }) + + with patch.object( + analytics.BlockingConnectionPool, + "from_url", + side_effect=ValueError("invalid URL"), + ): + assert await analytics.report_error(request, RuntimeError("boom")) is False From 09ed5c412b18e216add36f5657d5cce8980f4942 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:31:39 +0300 Subject: [PATCH 5/8] Redact credentials from analytics errors Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 14 +++++++++++++- tests/test_error_analytics.py | 13 +++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/api/analytics.py b/api/analytics.py index 3cbf53d9..488aae0d 100644 --- a/api/analytics.py +++ b/api/analytics.py @@ -2,6 +2,7 @@ import logging import os +import re from fastapi import Request from falkordb.asyncio import FalkorDB @@ -9,6 +10,17 @@ LOGGER = logging.getLogger(__name__) ANALYTICS_GRAPH = os.getenv("ORGANIZATIONS_GRAPH", "Organizations") +_SENSITIVE_VALUE = re.compile( + r"(?i)(password|token|secret|api[_-]?key|authorization)" + r"(\s*[=:]\s*|\s+)([^\s,;]+)" +) + + +def _safe_message(exc: Exception) -> str: + """Redact common credential forms before persisting an exception message.""" + message = _SENSITIVE_VALUE.sub(r"\1\2[REDACTED]", str(exc)) + message = re.sub(r"(?i)(redis(?:s)?://[^:@/\s]+:)[^@/\s]+@", r"\1[REDACTED]@", message) + return message[:4000] async def report_error(request: Request, exc: Exception) -> bool: @@ -42,7 +54,7 @@ async def report_error(request: Request, exc: Exception) -> bool: """, { "type": type(exc).__name__, - "message": str(exc)[:4000], + "message": _safe_message(exc), "endpoint": request.url.path, "method": request.method, "user_email": user_email, diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py index 6b526671..8c982168 100644 --- a/tests/test_error_analytics.py +++ b/tests/test_error_analytics.py @@ -10,6 +10,19 @@ pytestmark = [pytest.mark.unit] +def test_safe_message_redacts_credentials(): + """Persisted messages must not contain common credential values.""" + error = RuntimeError( + "password=hunter2 token: abc123 redis://default:secret@db.example:6379" + ) + + message = analytics._safe_message(error) # pylint: disable=protected-access + + assert "hunter2" not in message + assert "abc123" not in message + assert "secret@" not in message + + @pytest.mark.asyncio async def test_report_error_writes_org_graph(monkeypatch): """Unhandled errors should be written with QueryWeaver attribution.""" From de1698cbf623eed141b84cef2288a77da42b2906 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:32:19 +0300 Subject: [PATCH 6/8] Harden query error attribution Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 50 +++++++++++++++++++----------------- api/routes/graphs.py | 9 ++++++- api/routes/usage_tracking.py | 9 +++++-- tests/test_usage_tracking.py | 7 +++-- 4 files changed, 47 insertions(+), 28 deletions(-) diff --git a/api/analytics.py b/api/analytics.py index 488aae0d..3bafce06 100644 --- a/api/analytics.py +++ b/api/analytics.py @@ -3,6 +3,7 @@ import logging import os import re +import asyncio from fastapi import Request from falkordb.asyncio import FalkorDB @@ -36,29 +37,32 @@ async def report_error(request: Request, exc: Exception) -> bool: client = FalkorDB(connection_pool=pool) graph = client.select_graph(ANALYTICS_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, - }, + await asyncio.wait_for( + 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, + }, + ), + timeout=2, ) return True except Exception as analytics_error: # pylint: disable=broad-exception-caught diff --git a/api/routes/graphs.py b/api/routes/graphs.py index fdf9700f..8f2bc6d2 100644 --- a/api/routes/graphs.py +++ b/api/routes/graphs.py @@ -35,7 +35,9 @@ graphs_router = APIRouter(tags=["Graphs & Databases"]) -async def _serialize_pipeline(gen, *, user_id, namespaced, question, query_id): +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 @@ -65,6 +67,7 @@ async def _serialize_pipeline(gen, *, user_id, namespaced, question, query_id): question=question, error=final.error_message or "", query_id=query_id, + endpoint=endpoint, ) @@ -208,6 +211,7 @@ async def stream(): namespaced=namespaced, question=question, query_id=query_id, + endpoint=request.url.path, ): yield chunk except Exception: # pylint: disable=broad-exception-caught @@ -223,6 +227,7 @@ async def stream(): question=question, error="Unhandled streaming query failure", query_id=query_id, + endpoint=request.url.path, ) yield json.dumps({ "type": "error", @@ -269,6 +274,7 @@ async def stream(): namespaced=namespaced, question=question, query_id=query_id, + endpoint=request.url.path, ): yield chunk except Exception: # pylint: disable=broad-exception-caught @@ -282,6 +288,7 @@ async def stream(): 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 5966bfe1..eca7633b 100644 --- a/api/routes/usage_tracking.py +++ b/api/routes/usage_tracking.py @@ -58,7 +58,7 @@ source: 'queryweaver', type: 'QueryError', message: CASE WHEN $error = '' THEN 'Query could not be completed' ELSE $error END, - endpoint: '/graphs/' + $graph_id, + endpoint: $endpoint, method: 'POST', timestamp: timestamp() }) @@ -98,6 +98,7 @@ async def _write_usage( # pylint: disable=too-many-arguments,too-many-positiona success: bool, question: str, error: str, + endpoint: str, db, ) -> None: """Perform the single Cypher write against the Organizations graph.""" @@ -112,6 +113,7 @@ async def _write_usage( # pylint: disable=too-many-arguments,too-many-positiona "success": success, "question": question, "error": error, + "endpoint": endpoint, }, ) # Structured-ish log line so usage is visible to log aggregators even @@ -133,6 +135,7 @@ def record_query_usage_background( # pylint: disable=too-many-arguments,too-man question: str, error: str = "", query_id: Optional[str] = None, + endpoint: str = "", *, db=None, task_sink: Optional[set] = None, @@ -151,7 +154,8 @@ def record_query_usage_background( # pylint: disable=too-many-arguments,too-man 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 used to link an unhandled error. + 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 @@ -173,6 +177,7 @@ def record_query_usage_background( # pylint: disable=too-many-arguments,too-man success, question[:4000], error[:4000], + endpoint, db, ) ) diff --git a/tests/test_usage_tracking.py b/tests/test_usage_tracking.py index 6de84158..cfffb0da 100644 --- a/tests/test_usage_tracking.py +++ b/tests/test_usage_tracking.py @@ -83,6 +83,7 @@ async def test_records_successful_query_event(self): "success": True, "question": "How many users?", "error": "", + "endpoint": "", } @pytest.mark.asyncio @@ -97,7 +98,9 @@ async def test_records_failed_query_event(self): ) 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" @@ -163,7 +166,7 @@ def test_recorder_has_no_memory_or_provider_parameter(self): params = set(inspect.signature(record_query_usage_background).parameters) assert params == { "user_id", "namespaced", "success", "question", "error", "query_id", - "db", "task_sink" + "endpoint", "db", "task_sink" } assert "use_memory" not in params assert "provider" not in params From 641711387c1495e61bbcdf366f12543db50ef9c8 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:48:14 +0300 Subject: [PATCH 7/8] Address analytics privacy review Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 41 +++++----------- api/app_factory.py | 10 ++-- api/core/text2sql.py | 2 +- api/helpers/redaction.py | 30 ++++++++++++ api/routes/graphs.py | 2 +- api/routes/usage_tracking.py | 9 ++-- tests/test_error_analytics.py | 91 +++++++++++++++-------------------- tests/test_usage_tracking.py | 18 ++++++- 8 files changed, 111 insertions(+), 92 deletions(-) create mode 100644 api/helpers/redaction.py diff --git a/api/analytics.py b/api/analytics.py index 3bafce06..8da6e01f 100644 --- a/api/analytics.py +++ b/api/analytics.py @@ -1,41 +1,26 @@ """Best-effort error reporting to the organization analytics graph.""" -import logging -import os -import re import asyncio +import logging from fastapi import Request -from falkordb.asyncio import FalkorDB -from redis.asyncio import BlockingConnectionPool + +from api.config import ORGANIZATIONS_GRAPH +from api.extensions import db +from api.helpers.redaction import redact_sensitive_text LOGGER = logging.getLogger(__name__) -ANALYTICS_GRAPH = os.getenv("ORGANIZATIONS_GRAPH", "Organizations") -_SENSITIVE_VALUE = re.compile( - r"(?i)(password|token|secret|api[_-]?key|authorization)" - r"(\s*[=:]\s*|\s+)([^\s,;]+)" -) def _safe_message(exc: Exception) -> str: """Redact common credential forms before persisting an exception message.""" - message = _SENSITIVE_VALUE.sub(r"\1\2[REDACTED]", str(exc)) - message = re.sub(r"(?i)(redis(?:s)?://[^:@/\s]+:)[^@/\s]+@", r"\1[REDACTED]@", message) - return message[:4000] + return redact_sensitive_text(str(exc)) async def report_error(request: Request, exc: Exception) -> bool: """Record an unhandled QueryWeaver error without affecting the response.""" - url = os.getenv("FALKORDB_URL") - if not url: - LOGGER.error("Cannot report QueryWeaver error: FALKORDB_URL is not configured") - return False - - pool = None try: - pool = BlockingConnectionPool.from_url(url, decode_responses=True) - client = FalkorDB(connection_pool=pool) - graph = client.select_graph(ANALYTICS_GRAPH) + graph = db.select_graph(ORGANIZATIONS_GRAPH) user_email = getattr(request.state, "user_email", None) await asyncio.wait_for( graph.query( @@ -66,11 +51,9 @@ async def report_error(request: Request, exc: Exception) -> bool: ) return True except Exception as analytics_error: # pylint: disable=broad-exception-caught - LOGGER.error("Failed to report QueryWeaver error to analytics: %s", analytics_error) + LOGGER.error( + "Failed to report QueryWeaver error to analytics: %s", + analytics_error, + exc_info=True, + ) return False - finally: - if pool is not None: - try: - await pool.aclose() - except Exception: # pylint: disable=broad-exception-caught - LOGGER.debug("Failed to close analytics Redis pool", exc_info=True) diff --git a/api/app_factory.py b/api/app_factory.py index 060b4449..16d81491 100644 --- a/api/app_factory.py +++ b/api/app_factory.py @@ -330,6 +330,11 @@ async def handle_oauth_error( request: Request, exc: Exception ): # pylint: disable=unused-argument """Handle OAuth-related errors gracefully""" + await 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(): @@ -340,11 +345,6 @@ async def handle_oauth_error( if isinstance(exc, HTTPException): raise exc - await report_error(request, exc) - logging.exception( - "Unhandled error for %s %s", request.method, request.url.path, exc_info=exc - ) - # Preserve the existing FastAPI 500 response behavior. raise exc 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..af4735aa --- /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 8f2bc6d2..38003faf 100644 --- a/api/routes/graphs.py +++ b/api/routes/graphs.py @@ -265,7 +265,7 @@ async def confirm_destructive_operation( return JSONResponse(content={"error": "Invalid confirmation request"}, status_code=400) async def stream(): - question = confirm_data.chat[-1] if confirm_data.chat else "" + question = str(confirm_data.chat[-1]) if confirm_data.chat else "" query_id = str(uuid.uuid4()) try: async for chunk in _serialize_pipeline( diff --git a/api/routes/usage_tracking.py b/api/routes/usage_tracking.py index eca7633b..b574eceb 100644 --- a/api/routes/usage_tracking.py +++ b/api/routes/usage_tracking.py @@ -32,6 +32,7 @@ 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 @@ -53,11 +54,11 @@ error: $error, timestamp: timestamp() }) -FOREACH (_ IN CASE WHEN $success THEN [] ELSE [1] END | +FOREACH (_ IN CASE WHEN $error = '' THEN [] ELSE [1] END | CREATE (error:Error { source: 'queryweaver', type: 'QueryError', - message: CASE WHEN $error = '' THEN 'Query could not be completed' ELSE $error END, + message: $error, endpoint: $endpoint, method: 'POST', timestamp: timestamp() @@ -175,8 +176,8 @@ def record_query_usage_background( # pylint: disable=too-many-arguments,too-man namespaced, is_demo, success, - question[:4000], - error[:4000], + redact_sensitive_text(str(question)), + redact_sensitive_text(error), endpoint, db, ) diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py index 8c982168..ff563f85 100644 --- a/tests/test_error_analytics.py +++ b/tests/test_error_analytics.py @@ -10,47 +10,57 @@ 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 redis://default:secret@db.example:6379" + 'password=hunter2 token: abc123 Authorization: ****** ' + '"api_key": "json-secret" ******db.example:5432/app ' + '******host/db' ) message = analytics._safe_message(error) # pylint: disable=protected-access - assert "hunter2" not in message - assert "abc123" not in message - assert "secret@" not in message + for secret in ( + "hunter2", "abc123", "bearer-secret", "json-secret", + "pg-secret", "mysql-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(monkeypatch): +async def test_report_error_writes_org_graph(): """Unhandled errors should be written with QueryWeaver attribution.""" - monkeypatch.setenv("FALKORDB_URL", "redis://analytics.example:6379") graph = AsyncMock() client = MagicMock() client.select_graph.return_value = graph - pool = AsyncMock() - scope = { - "type": "http", - "method": "POST", - "path": "/graphs/query", - "headers": [], - "query_string": b"", - "scheme": "https", - "server": ("queryweaver.example", 443), - } - request = Request(scope) - request.state.user_email = "user@example.com" - with ( - patch.object(analytics.BlockingConnectionPool, "from_url", return_value=pool), - patch.object(analytics, "FalkorDB", return_value=client), - ): - result = await analytics.report_error(request, RuntimeError("sales demo failed")) + with patch.object(analytics, "db", client): + result = await analytics.report_error(_request(), RuntimeError("sales demo failed")) assert result is True - graph.query.assert_awaited_once() params = graph.query.await_args.args[1] assert params == { "type": "RuntimeError", @@ -59,33 +69,12 @@ async def test_report_error_writes_org_graph(monkeypatch): "method": "POST", "user_email": "user@example.com", } - pool.aclose.assert_awaited_once() - - -@pytest.mark.asyncio -async def test_report_error_without_url_returns_false(monkeypatch): - """Missing analytics configuration must not mask the original error.""" - monkeypatch.delenv("FALKORDB_URL", raising=False) - request = Request({ - "type": "http", "method": "GET", "path": "/", "headers": [], - "query_string": b"", "scheme": "http", "server": ("localhost", 80), - }) - - assert await analytics.report_error(request, RuntimeError("boom")) is False @pytest.mark.asyncio -async def test_report_error_swallows_setup_failure(monkeypatch): - """Invalid connection configuration must preserve best-effort behavior.""" - monkeypatch.setenv("FALKORDB_URL", "invalid") - request = Request({ - "type": "http", "method": "GET", "path": "/", "headers": [], - "query_string": b"", "scheme": "http", "server": ("localhost", 80), - }) - - with patch.object( - analytics.BlockingConnectionPool, - "from_url", - side_effect=ValueError("invalid URL"), - ): - assert await analytics.report_error(request, RuntimeError("boom")) is False +async def test_report_error_swallows_setup_failure(): + """Analytics connection failures must preserve best-effort behavior.""" + broken_db = MagicMock() + broken_db.select_graph.side_effect = ConnectionError("offline") + with patch.object(analytics, "db", broken_db): + assert await analytics.report_error(_request(), RuntimeError("boom")) is False diff --git a/tests/test_usage_tracking.py b/tests/test_usage_tracking.py index cfffb0da..1f1a5663 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 @@ -75,9 +76,9 @@ async def test_records_successful_query_event(self): 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, - "query_id": params["query_id"], "graph_id": f"{USER_ID}_mydb", "is_demo": False, "success": True, @@ -105,6 +106,21 @@ async def test_records_failed_query_event(self): assert params["question"] == "Broken query" assert params["error"] == "syntax error" + @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): db, graph = _mock_db() From d7ab4fb2f063f9658d955d146ef00850aad251d5 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:23:56 +0300 Subject: [PATCH 8/8] Address follow-up analytics review Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- api/analytics.py | 114 ++++++++++++++++++++++------------ api/app_factory.py | 2 +- api/helpers/redaction.py | 2 +- api/routes/usage_tracking.py | 8 ++- tests/test_error_analytics.py | 22 +++++-- tests/test_usage_tracking.py | 1 + 6 files changed, 98 insertions(+), 51 deletions(-) diff --git a/api/analytics.py b/api/analytics.py index 8da6e01f..1dcdbed6 100644 --- a/api/analytics.py +++ b/api/analytics.py @@ -2,14 +2,15 @@ import asyncio import logging +from typing import Optional from fastapi import Request from api.config import ORGANIZATIONS_GRAPH -from api.extensions import db from api.helpers.redaction import redact_sensitive_text LOGGER = logging.getLogger(__name__) +_DB_OVERRIDE = None def _safe_message(exc: Exception) -> str: @@ -17,43 +18,76 @@ def _safe_message(exc: Exception) -> str: return redact_sensitive_text(str(exc)) -async def report_error(request: Request, exc: Exception) -> bool: - """Record an unhandled QueryWeaver error without affecting the response.""" - try: - graph = db.select_graph(ORGANIZATIONS_GRAPH) - user_email = getattr(request.state, "user_email", None) - await asyncio.wait_for( - 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, - }, - ), - timeout=2, - ) - return True - except Exception as analytics_error: # pylint: disable=broad-exception-caught - LOGGER.error( - "Failed to report QueryWeaver error to analytics: %s", - analytics_error, - exc_info=True, +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) ) - return False + """, + { + "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 16d81491..b338e197 100644 --- a/api/app_factory.py +++ b/api/app_factory.py @@ -330,7 +330,7 @@ async def handle_oauth_error( request: Request, exc: Exception ): # pylint: disable=unused-argument """Handle OAuth-related errors gracefully""" - await report_error(request, exc) + report_error(request, exc) logging.exception( "Unhandled error for %s %s", request.method, request.url.path ) diff --git a/api/helpers/redaction.py b/api/helpers/redaction.py index af4735aa..5e49e80c 100644 --- a/api/helpers/redaction.py +++ b/api/helpers/redaction.py @@ -15,7 +15,7 @@ """ ) _CONNECTION_PASSWORD = re.compile( - r"(?i)\b([a-z][a-z0-9+.\-]*://[^:@/\s]+:)[^@/\s]+@" + r"(?i)\b([a-z][a-z0-9+.\-]*://[^:@/\s]*:)[^@/\s]+@" ) diff --git a/api/routes/usage_tracking.py b/api/routes/usage_tracking.py index b574eceb..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 @@ -42,7 +44,7 @@ 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 { diff --git a/tests/test_error_analytics.py b/tests/test_error_analytics.py index ff563f85..5175a430 100644 --- a/tests/test_error_analytics.py +++ b/tests/test_error_analytics.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, MagicMock, patch +import asyncio import pytest from fastapi import Request @@ -29,7 +30,8 @@ def test_safe_message_redacts_credentials(): error = RuntimeError( 'password=hunter2 token: abc123 Authorization: ****** ' '"api_key": "json-secret" ******db.example:5432/app ' - '******host/db' + '******host/db ' + 'redis://:redis-secret@cache.example:6379' ) message = analytics._safe_message(error) # pylint: disable=protected-access @@ -37,6 +39,7 @@ def test_safe_message_redacts_credentials(): for secret in ( "hunter2", "abc123", "bearer-secret", "json-secret", "pg-secret", "mysql-secret", + "redis-secret", ): assert secret not in message @@ -57,10 +60,13 @@ async def test_report_error_writes_org_graph(): client = MagicMock() client.select_graph.return_value = graph - with patch.object(analytics, "db", client): - result = await analytics.report_error(_request(), RuntimeError("sales demo failed")) + 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)) - assert result is True params = graph.query.await_args.args[1] assert params == { "type": "RuntimeError", @@ -74,7 +80,11 @@ async def test_report_error_writes_org_graph(): @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", broken_db): - assert await analytics.report_error(_request(), RuntimeError("boom")) is False + 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 1f1a5663..c9f22f55 100644 --- a/tests/test_usage_tracking.py +++ b/tests/test_usage_tracking.py @@ -105,6 +105,7 @@ async def test_records_failed_query_event(self): 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):