diff --git a/CHANGELOG.md b/CHANGELOG.md index 922b766..f7f0180 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `/ingest/status/{job_id}` is backed by the Redis `JobManager` (jobs are tenant-scoped; other tenants' jobs return 404). `/ingest/page`, `/ingest/batch` and `/ingest/all` now record job entries and return a `job_id`. - `/ingest/repo-status/{repository}` reports wiki page count vs indexed Document-node count under `users/{tenant}/{repository}` plus the tenant's Redis job statistics. - `/deduplicate/check` runs a tenant-scoped Qdrant similarity scan: wiki chunk pairs above the threshold (default 0.9 cosine) grouped per page pair with best score, matching chunk-pair count, and page references. Read-only. +- **Nightly integrity check** — `POST /maintenance/integrity-check {user}` (read-only: reports, never auto-fixes) reports per tenant: wiki pages with ZERO vectors in Qdrant (silent-skip reindex victims), orphaned vectors whose wiki page no longer exists, unexpected Qdrant collections (test-tenant residue and unknown namespaces flagged; other services' collections counted as foreign), Neo4j Document nodes without wiki counterparts, plus counts and duration. The latest report is cached in Redis (30 days) so the weekly quality report can fold it in. ### Changed diff --git a/src/routers/maintenance.py b/src/routers/maintenance.py index 33b26d2..2c87742 100644 --- a/src/routers/maintenance.py +++ b/src/routers/maintenance.py @@ -21,6 +21,7 @@ from src.core.dependencies import ( VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep, QdrantDep, OllamaDep, PaperlessDep, verify_api_key ) +from src.core.multi_tenancy import RequiredUser, sanitize_user_id from src.config import get_settings from datetime import datetime, timezone @@ -1098,3 +1099,246 @@ async def reconcile_index( except Exception as e: logger.error(f"Reconcile-index failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) + + +# ========== Integrity check (nightly, read-only) ========== + +# Redis key holding the latest integrity report per tenant (folded into the +# weekly quality report). +INTEGRITY_LATEST_KEY = "library:integrity:latest:{user}" +INTEGRITY_LATEST_TTL = 86400 * 30 # 30 days + +#: Qdrant collection prefixes owned by library-desk. +LIBRARY_COLLECTION_PREFIXES = ("library_desk_", "volatile_") + + +def _looks_like_test_tenant(name: str) -> bool: + """Heuristic for test/probe residue in collection or tenant names.""" + lowered = name.lower() + return ( + "llm_tester" in lowered + or "llm-tester" in lowered + or "test" in lowered + or lowered.startswith("verify_probe") + or lowered.startswith("verify-probe") + ) + + +def classify_collection(name: str, known_tenants: set[str]) -> str: + """ + Classify a Qdrant collection against known tenant patterns. + + Returns one of: + - ``expected``: library-desk collection for a tenant with a wiki namespace + - ``test_residue``: library-desk collection for a test/probe tenant + - ``unknown_tenant``: library-desk collection for a tenant with no wiki + namespace (orphaned or mis-scoped) + - ``foreign_test_residue``: another service's collection that looks like + test residue (reported, but owned elsewhere) + - ``foreign``: another service's collection (informational only) + """ + for prefix in LIBRARY_COLLECTION_PREFIXES: + if name.startswith(prefix): + tenant = name[len(prefix):] + if _looks_like_test_tenant(tenant): + return "test_residue" + if tenant in known_tenants: + return "expected" + return "unknown_tenant" + if _looks_like_test_tenant(name): + return "foreign_test_residue" + return "foreign" + + +class IntegrityCheckRequest(BaseModel): + """Request body for /maintenance/integrity-check.""" + user: RequiredUser = Field( + ..., + description="User identifier (tenant). Required — the report is scoped to this tenant." + ) + + +class IntegrityCheckResponse(BaseModel): + """Read-only integrity report for one tenant.""" + success: bool + user: str + generated_at: str + pages_without_vectors: List[Dict[str, Any]] = Field( + default_factory=list, + description="Wiki pages with ZERO vectors in Qdrant (silent-skip reindex victims)" + ) + orphaned_vector_chunks: int = Field( + default=0, description="Vector chunks whose wiki page no longer exists" + ) + orphaned_vector_page_ids: List[int] = Field( + default_factory=list, description="Distinct stale page ids referenced by orphaned chunks" + ) + unexpected_collections: List[Dict[str, str]] = Field( + default_factory=list, + description="Qdrant collections flagged as test residue or unknown tenants" + ) + foreign_collections: int = Field( + default=0, description="Collections owned by other services (informational)" + ) + documents_without_wiki: List[Dict[str, Any]] = Field( + default_factory=list, + description="Neo4j Document nodes whose wiki page no longer exists" + ) + counts: Dict[str, int] = Field(default_factory=dict) + duration_ms: float = 0.0 + + +async def run_integrity_check( + user: str, + vector_service: VectorService, + graph_service: GraphService, + wiki_client, + qdrant +) -> IntegrityCheckResponse: + """ + Run the read-only integrity check for one tenant. + + Reports (never fixes): + 1. Wiki pages with zero vectors in the tenant's Qdrant collection + 2. Orphaned vectors whose wiki page no longer exists + 3. Unexpected Qdrant collections (test residue / unknown tenants) + 4. Neo4j Document nodes without wiki counterparts + """ + start_time = time.time() + tenant_prefix = f"users/{sanitize_user_id(user)}" + + # One unfiltered listing serves both the tenant scan and the + # known-tenant derivation for collection classification. + all_pages = await wiki_client.list_all_pages() + tenant_pages = [ + p for p in all_pages + if ("/" + str(p.get("path", "")).lstrip("/")).startswith("/" + tenant_prefix) + ] + + known_tenants = set() + for p in all_pages: + parts = str(p.get("path", "")).lstrip("/").split("/") + if len(parts) >= 2 and parts[0] == "users": + known_tenants.add(sanitize_user_id(parts[1])) + + tenant_page_ids = {p["id"] for p in tenant_pages if p.get("id")} + + # Vector side (tenant collection only) + chunk_refs = await vector_service.get_all_chunk_references(user) + wiki_chunk_refs = [r for r in chunk_refs if r.get("doc_type", "wiki") == "wiki"] + vectorized_page_ids = {r["page_id"] for r in wiki_chunk_refs if r.get("page_id")} + + pages_without_vectors = [ + {"page_id": p["id"], "path": p.get("path", ""), "title": p.get("title", "")} + for p in tenant_pages + if p.get("id") and p["id"] not in vectorized_page_ids + ] + + orphaned_chunks = [ + r for r in wiki_chunk_refs + if r.get("page_id") and r["page_id"] not in tenant_page_ids + ] + orphaned_page_ids = sorted({r["page_id"] for r in orphaned_chunks}) + + # Collection audit (global listing, read-only) + collections = await qdrant.list_collections() + unexpected = [] + foreign_count = 0 + for coll in collections: + category = classify_collection(coll["name"], known_tenants) + if category in ("test_residue", "unknown_tenant", "foreign_test_residue"): + unexpected.append({"name": coll["name"], "category": category}) + elif category == "foreign": + foreign_count += 1 + + # Graph side (tenant labels only) + graph_docs = await graph_service.get_all_document_references(user) + documents_without_wiki = [ + {"page_id": d.get("page_id"), "path": d.get("path", ""), "title": d.get("title", "")} + for d in graph_docs + if d.get("doc_type") == "wiki" + and d.get("page_id") + and d["page_id"] not in tenant_page_ids + ] + + duration_ms = (time.time() - start_time) * 1000 + + return IntegrityCheckResponse( + success=True, + user=user, + generated_at=datetime.now(timezone.utc).isoformat(), + pages_without_vectors=pages_without_vectors, + orphaned_vector_chunks=len(orphaned_chunks), + orphaned_vector_page_ids=orphaned_page_ids, + unexpected_collections=unexpected, + foreign_collections=foreign_count, + documents_without_wiki=documents_without_wiki, + counts={ + "tenant_wiki_pages": len(tenant_pages), + "tenant_vector_chunks": len(wiki_chunk_refs), + "tenant_graph_documents": len(graph_docs), + "pages_without_vectors": len(pages_without_vectors), + "orphaned_vector_chunks": len(orphaned_chunks), + "unexpected_collections": len(unexpected), + "documents_without_wiki": len(documents_without_wiki), + }, + duration_ms=duration_ms + ) + + +@router.post("/integrity-check", response_model=IntegrityCheckResponse) +async def integrity_check( + request: IntegrityCheckRequest, + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + qdrant: QdrantDep = None, + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Nightly integrity check (READ-ONLY: reports, never auto-fixes). + + Reports per tenant: + - Wiki pages with ZERO vectors in Qdrant (silent-skip reindex victims) + - Orphaned vectors whose wiki page no longer exists + - Unexpected Qdrant collections: anything not matching known tenant + patterns — flags test-tenant residue and unknown namespaces + - Neo4j Document nodes without wiki counterparts + - Counts and duration + + The latest report is cached in Redis (30 days) so the weekly quality + report can fold it in without re-running the scan. + + **Scheduler Task** — nightly at 04:30, see docs/scheduler-tasks.md. + """ + try: + report = await run_integrity_check( + user=request.user, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + qdrant=qdrant + ) + + # Cache the latest report for the quality report (best-effort) + if redis: + try: + import json as _json + await redis.setex( + INTEGRITY_LATEST_KEY.format(user=request.user), + INTEGRITY_LATEST_TTL, + _json.dumps(report.model_dump(mode="json")) + ) + except Exception as e: + logger.warning(f"Failed to cache integrity report: {e}") + + logger.info( + f"Integrity check for {request.user}: {report.counts} " + f"in {report.duration_ms:.0f}ms" + ) + return report + + except Exception as e: + logger.error(f"Integrity check failed for {request.user}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Integrity check failed") diff --git a/tests/test_integrity_check.py b/tests/test_integrity_check.py new file mode 100644 index 0000000..af70e9c --- /dev/null +++ b/tests/test_integrity_check.py @@ -0,0 +1,186 @@ +""" +Offline unit tests for the nightly integrity-check endpoint (Phase C item 2). + +All external clients are mocked - no shared services are contacted. +The endpoint must be strictly read-only: these tests assert that no +delete/purge/update methods are ever invoked. +""" + +import pytest +from unittest.mock import AsyncMock + +from src.routers.maintenance import ( + IntegrityCheckRequest, + classify_collection, + integrity_check, + run_integrity_check, +) + +TEST_USER = "llm_tester" +TENANT_PREFIX = f"users/{TEST_USER}" + + +def _mock_clients( + wiki_pages, + chunk_refs, + graph_docs, + collections, +): + wiki_client = AsyncMock() + wiki_client.list_all_pages = AsyncMock(return_value=wiki_pages) + + vector_service = AsyncMock() + vector_service.get_all_chunk_references = AsyncMock(return_value=chunk_refs) + + graph_service = AsyncMock() + graph_service.get_all_document_references = AsyncMock(return_value=graph_docs) + + qdrant = AsyncMock() + qdrant.list_collections = AsyncMock(return_value=collections) + + return wiki_client, vector_service, graph_service, qdrant + + +class TestRunIntegrityCheck: + @pytest.mark.asyncio + async def test_full_report(self): + wiki_pages = [ + # tenant pages + {"id": 1, "path": f"{TENANT_PREFIX}/a", "title": "A"}, + {"id": 2, "path": f"{TENANT_PREFIX}/b", "title": "B"}, + # another tenant's page (defines a known tenant, out of scope here) + {"id": 50, "path": "users/jpmschweitzer/x", "title": "X"}, + ] + chunk_refs = [ + # page 1 has vectors; page 2 has none (silent-skip victim) + {"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"}, + # orphan: page 99 no longer exists in the wiki + {"chunk_id": "c9", "page_id": 99, "doc_type": "wiki"}, + # non-wiki chunk is ignored by the wiki-side checks + {"chunk_id": "cd", "document_id": "d1", "doc_type": "document"}, + ] + graph_docs = [ + {"page_id": 1, "path": f"{TENANT_PREFIX}/a", "title": "A", "doc_type": "wiki"}, + # stale Document node: wiki page 77 is gone + {"page_id": 77, "path": f"{TENANT_PREFIX}/old", "title": "Old", "doc_type": "wiki"}, + ] + collections = [ + {"name": "library_desk_jpmschweitzer", "vectors_count": 10}, + {"name": "library_desk_llm_tester", "vectors_count": 2}, + {"name": "library_desk_ghost_tenant", "vectors_count": 1}, + {"name": "open-webui_files", "vectors_count": 5}, + ] + + wiki_client, vector_service, graph_service, qdrant = _mock_clients( + wiki_pages, chunk_refs, graph_docs, collections + ) + + report = await run_integrity_check( + TEST_USER, vector_service, graph_service, wiki_client, qdrant + ) + + assert report.success is True + assert report.user == TEST_USER + + # Pages without vectors: page 2 only (page 1 has c1) + assert [p["page_id"] for p in report.pages_without_vectors] == [2] + + # Orphaned vectors: c9 -> page 99 + assert report.orphaned_vector_chunks == 1 + assert report.orphaned_vector_page_ids == [99] + + # Collections: llm_tester is test residue, ghost_tenant unknown, + # jpmschweitzer expected, open-webui foreign + flagged = {c["name"]: c["category"] for c in report.unexpected_collections} + assert flagged == { + "library_desk_llm_tester": "test_residue", + "library_desk_ghost_tenant": "unknown_tenant", + } + assert report.foreign_collections == 1 + + # Graph documents without wiki counterparts: page 77 + assert [d["page_id"] for d in report.documents_without_wiki] == [77] + + assert report.counts["tenant_wiki_pages"] == 2 + assert report.counts["pages_without_vectors"] == 1 + assert report.counts["documents_without_wiki"] == 1 + assert report.duration_ms >= 0 + + @pytest.mark.asyncio + async def test_read_only_no_mutations(self): + """The integrity check must never call any destructive method.""" + wiki_client, vector_service, graph_service, qdrant = _mock_clients( + [], [], [], [] + ) + + await run_integrity_check( + TEST_USER, vector_service, graph_service, wiki_client, qdrant + ) + + for mock, destructive in ( + (vector_service, ("purge_chunks_by_ids", "delete_page_chunks")), + (graph_service, ("purge_orphan_entities", "purge_stale_documents_by_ids", + "delete_page", "cleanup_broken_relationships")), + (qdrant, ("delete_collection", "delete_by_ids", "delete_by_filter")), + (wiki_client, ("delete_page", "update_page", "create_page")), + ): + for name in destructive: + assert not getattr(mock, name).await_count, ( + f"integrity check must be read-only but called {name}" + ) + + +class TestIntegrityEndpoint: + @pytest.mark.asyncio + async def test_caches_latest_report_in_redis(self): + wiki_client, vector_service, graph_service, qdrant = _mock_clients( + [{"id": 1, "path": f"{TENANT_PREFIX}/a", "title": "A"}], + [{"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"}], + [], + [], + ) + redis = AsyncMock() + + report = await integrity_check( + request=IntegrityCheckRequest(user=TEST_USER), + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + qdrant=qdrant, + redis=redis, + api_key="", + ) + + assert report.success is True + redis.setex.assert_awaited_once() + key = redis.setex.await_args.args[0] + assert key == f"library:integrity:latest:{TEST_USER}" + + def test_requires_user(self): + with pytest.raises(Exception): + IntegrityCheckRequest(user=" ") + + +class TestClassifyCollection: + def test_expected_tenant(self): + assert classify_collection( + "library_desk_jpmschweitzer", {"jpmschweitzer"} + ) == "expected" + + def test_test_residue_beats_known_tenant(self): + # Even if the test tenant has wiki pages during a run, its + # collections are still flagged as residue. + assert classify_collection( + "library_desk_llm_tester", {"llm_tester"} + ) == "test_residue" + + def test_unknown_tenant(self): + assert classify_collection( + "volatile_mystery", {"jpmschweitzer"} + ) == "unknown_tenant" + + def test_foreign_and_foreign_residue(self): + assert classify_collection("open-webui_files", set()) == "foreign" + assert classify_collection("core_ai_user_test_at_example_com", set()) == \ + "foreign_test_residue" + assert classify_collection("test_user", set()) == "foreign_test_residue"