fix(storage): harden bloat diagnostic evidence

Author: alteixeira20

routes/diagnostics_routes.py

@@ -1,5 +1,6 @@
 """Diagnostics routes — /api/db/stats, /api/rag/stats, /api/test/youtube, /api/test-research."""
 
+import asyncio
 import logging
 import os
 from typing import Dict, Any
@@ -58,7 +59,8 @@ async def get_diagnostics_logs(request: Request, limit: int = 200) -> Dict[str,
     async def get_storage_bloat_diagnostics(request: Request) -> Dict[str, Any]:
         require_admin(request)
         try:
-            return collect_configured_storage_bloat_diagnostics(
+            return await asyncio.to_thread(
+                collect_configured_storage_bloat_diagnostics,
                 default_db_path=APP_DB,
                 upload_dir=UPLOAD_DIR,
             )

src/storage_diagnostics.py

@@ -49,6 +49,28 @@
     "extra",
     "attachments",
 )
+REFERENCE_COVERED_SOURCES = (
+    "chat_messages.content",
+    "chat_messages.attachment_metadata",
+    "documents.current_content.pdf_source_marker",
+    "document_versions.content.pdf_source_marker",
+)
+REFERENCE_UNCOVERED_SOURCES = (
+    "gallery_images.filename",
+    "gallery_images.file_hash",
+    "notes.image_url",
+    "notes.color",
+    "notes.content",
+    "notes.items",
+    "calendars.color",
+    "calendar_events.color",
+    "calendar_events.description",
+    "calendar_events.location",
+)
+DURABLE_REFERENCE_SOURCES_INCOMPLETE = "durable_reference_sources_incomplete"
+INCOMPLETE_REFERENCE_SAMPLE_SCOPE = (
+    "candidates_observed_within_incomplete_reference_scope"
+)
 REFERENCE_CONTAINER_KEYS = {
     "attachment",
     "attachments",
@@ -96,7 +118,8 @@ def collect_storage_bloat_diagnostics(
     upload_report = _collect_upload_report(
         Path(upload_dir or UPLOAD_DIR),
         live_upload_ids,
-        db_report["upload_references"]["complete"],
+        db_report["upload_references"]["attempted_scan_complete"],
+        db_report["upload_references"]["reference_coverage"]["complete"],
         warnings,
     )
     return {
@@ -201,6 +224,15 @@ def _connect_readonly(db_path: Path) -> sqlite3.Connection:
     return conn
 
 
+def _durable_reference_coverage_report() -> dict[str, Any]:
+    """Describe implemented reference scans without exposing persisted values."""
+    return {
+        "complete": not REFERENCE_UNCOVERED_SOURCES,
+        "covered_sources": list(REFERENCE_COVERED_SOURCES),
+        "uncovered_sources": list(REFERENCE_UNCOVERED_SOURCES),
+    }
+
+
 def _collect_db_report(
     db_path: Path | None,
     warnings: list[str],
@@ -221,7 +253,9 @@ def _collect_db_report(
             "scan_limit_scope": "per_source",
             "value_char_scan_limit": MAX_REFERENCE_CONTENT_CHARS,
             "json_node_scan_limit": MAX_REFERENCE_JSON_NODES,
+            "attempted_scan_complete": False,
             "complete": False,
+            "reference_coverage": _durable_reference_coverage_report(),
         },
     }
     live_upload_ids: set[str] = set()
@@ -491,7 +525,9 @@ def _live_upload_reference_report(
         "scan_limit_scope": "per_source",
         "value_char_scan_limit": MAX_REFERENCE_CONTENT_CHARS,
         "json_node_scan_limit": MAX_REFERENCE_JSON_NODES,
+        "attempted_scan_complete": False,
         "complete": False,
+        "reference_coverage": _durable_reference_coverage_report(),
     }
     live_ids: set[str] = set()
     scans_complete: list[bool] = []
@@ -575,7 +611,11 @@ def _live_upload_reference_report(
             )
 
     report["live_reference_count"] = len(live_ids)
-    report["complete"] = bool(scans_complete) and all(scans_complete)
+    attempted_scan_complete = bool(scans_complete) and all(scans_complete)
+    report["attempted_scan_complete"] = attempted_scan_complete
+    report["complete"] = (
+        attempted_scan_complete and report["reference_coverage"]["complete"]
+    )
     return report, live_ids
 
 
@@ -788,6 +828,7 @@ def _collect_upload_report(
     upload_dir: Path,
     live_upload_ids: set[str],
     reference_scan_complete: bool,
+    durable_reference_coverage_complete: bool,
     warnings: list[str],
 ) -> dict[str, Any]:
     report: dict[str, Any] = {
@@ -813,6 +854,9 @@ def _collect_upload_report(
             observed_count=0,
             sample=[],
             reference_scan_complete=reference_scan_complete,
+            durable_reference_coverage_complete=(
+                durable_reference_coverage_complete
+            ),
             upload_traversal_complete=False,
         ),
     }
@@ -831,6 +875,7 @@ def _collect_upload_report(
         manifest_ids,
         live_upload_ids,
         reference_scan_complete,
+        durable_reference_coverage_complete,
         warnings,
     )
     report.update(file_report)
@@ -845,6 +890,7 @@ def _upload_file_report(
     manifest_ids: set[str],
     live_upload_ids: set[str],
     reference_scan_complete: bool,
+    durable_reference_coverage_complete: bool,
     warnings: list[str],
 ) -> dict[str, Any]:
     state = UploadTraversalState(visited_dirs=1)
@@ -861,7 +907,7 @@ def _upload_file_report(
         orphan_count += 1
         if len(orphan_sample) < MAX_ORPHAN_ROWS:
             orphan_sample.append(
-                _upload_sample(record, upload_dir, manifest_ids, live_upload_ids)
+                _upload_sample(record, upload_dir, manifest_ids)
             )
 
     traversal_unreadable = state.skipped_unreadable_count > 0
@@ -884,6 +930,9 @@ def _upload_file_report(
             observed_count=orphan_count,
             sample=orphan_sample,
             reference_scan_complete=reference_scan_complete,
+            durable_reference_coverage_complete=(
+                durable_reference_coverage_complete
+            ),
             upload_traversal_complete=not traversal_incomplete,
             upload_traversal_truncated=state.truncated,
             upload_traversal_unreadable=traversal_unreadable,
@@ -896,13 +945,16 @@ def _suspected_orphan_report(
     observed_count: int,
     sample: list[dict[str, Any]],
     reference_scan_complete: bool,
+    durable_reference_coverage_complete: bool,
     upload_traversal_complete: bool,
     upload_traversal_truncated: bool = False,
     upload_traversal_unreadable: bool = False,
 ) -> dict[str, Any]:
     reasons: list[str] = []
     if not reference_scan_complete:
         reasons.append("db_reference_scan_incomplete")
+    if not durable_reference_coverage_complete:
+        reasons.append(DURABLE_REFERENCE_SOURCES_INCOMPLETE)
     if upload_traversal_truncated or (
         not upload_traversal_complete and not upload_traversal_unreadable
     ):
@@ -917,6 +969,11 @@ def _suspected_orphan_report(
         "count": observed_count if complete else None,
         "observed_count": observed_count,
         "reason_incomplete": reasons or None,
+        "sample_scope": (
+            "suspected_orphan_candidates"
+            if complete
+            else INCOMPLETE_REFERENCE_SAMPLE_SCOPE
+        ),
         "sample": sample,
     }
 
@@ -1024,14 +1081,12 @@ def _upload_sample(
     record: UploadFileRecord,
     upload_dir: Path,
     manifest_ids: set[str],
-    live_upload_ids: set[str],
 ) -> dict[str, Any]:
     return {
         "path_hash": _hash_upload_identifier(_relative_upload_path(upload_dir, record.path)),
         "relative_dir_bucket": _safe_relative_dir_bucket(upload_dir, record.path),
         "size_bytes": record.size_bytes,
         "manifest_present": _upload_file_in_manifest(record.path, manifest_ids),
-        "live_db_reference_found": _upload_file_is_live(record.path, live_upload_ids),
     }
 
 

tests/test_diagnostics_service_route.py

@@ -1,4 +1,4 @@
-"""Route-level regression tests for GET /api/diagnostics/services.
+"""Route-level regression tests for admin diagnostics endpoints.
 
 The reviewer asked for explicit coverage of unauthenticated / non-admin / admin
 access to this admin diagnostics route, beyond the unit tests for the collector.
@@ -8,6 +8,9 @@
 fail, so the suite stays green in minimal environments; CI installs
 requirements, so the tests run there.
 """
+import asyncio
+import threading
+
 import pytest
 
 fastapi = pytest.importorskip("fastapi")
@@ -66,3 +69,95 @@ def gate(_request: Request):
     body = r.json()
     assert set(body) == {"overall", "services", "timestamp"}
     assert body["overall"] == "ok"
+
+
+def _storage_bloat_endpoint():
+    router = diag.setup_diagnostics_routes(
+        rag_manager=None,
+        rag_available=False,
+        research_handler=None,
+        memory_vector=None,
+    )
+    return next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/diagnostics/storage-bloat"
+    )
+
+
+@pytest.mark.asyncio
+async def test_storage_bloat_collection_leaves_event_loop_responsive(monkeypatch):
+    loop = asyncio.get_running_loop()
+    event_loop_thread = threading.get_ident()
+    collector_started = asyncio.Event()
+    collector_release = threading.Event()
+    collector_threads = []
+    expected = {
+        "status": "success",
+        "database": {},
+        "uploads": {},
+        "warnings": [],
+    }
+
+    def blocked_collector(*, default_db_path, upload_dir):
+        assert default_db_path == diag.APP_DB
+        assert upload_dir == diag.UPLOAD_DIR
+        collector_threads.append(threading.get_ident())
+        loop.call_soon_threadsafe(collector_started.set)
+        if not collector_release.wait(timeout=2):
+            raise AssertionError("collector was not released by the event loop")
+        return expected
+
+    monkeypatch.setattr(diag, "require_admin", lambda _request: None)
+    monkeypatch.setattr(
+        diag,
+        "collect_configured_storage_bloat_diagnostics",
+        blocked_collector,
+    )
+    endpoint = _storage_bloat_endpoint()
+    route_task = asyncio.create_task(endpoint(object()))
+
+    try:
+        await asyncio.wait_for(collector_started.wait(), timeout=2)
+        heartbeat = asyncio.Event()
+        loop.call_soon(heartbeat.set)
+        await asyncio.wait_for(heartbeat.wait(), timeout=2)
+
+        assert route_task.done() is False
+        assert len(collector_threads) == 1
+        assert collector_threads[0] != event_loop_thread
+    finally:
+        collector_release.set()
+
+    assert await asyncio.wait_for(route_task, timeout=2) == expected
+
+
+def test_storage_bloat_collector_error_returns_generic_500(monkeypatch):
+    def failing_collector(*, default_db_path, upload_dir):
+        raise RuntimeError("private database path and credentials")
+
+    monkeypatch.setattr(diag, "require_admin", lambda _request: None)
+    monkeypatch.setattr(
+        diag,
+        "collect_configured_storage_bloat_diagnostics",
+        failing_collector,
+    )
+    app = FastAPI()
+    app.include_router(
+        diag.setup_diagnostics_routes(
+            rag_manager=None,
+            rag_available=False,
+            research_handler=None,
+            memory_vector=None,
+        )
+    )
+    client = TestClient(app, raise_server_exceptions=False)
+
+    response = client.get("/api/diagnostics/storage-bloat")
+
+    assert response.status_code == 500
+    assert response.json() == {
+        "detail": "Failed to retrieve storage bloat diagnostics"
+    }
+    assert "private database path" not in response.text
+    assert "credentials" not in response.text