diff --git a/CHANGELOG.md b/CHANGELOG.md index 48941e6..922b766 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Stub endpoints implemented** (`/ingest/check-updates`, `/ingest/status/{job_id}`, `/ingest/repo-status/{repository}`, `/deduplicate/check`) — all previously returned canned "not yet implemented" responses; all now require an explicit `user` (Phase B rule): + - `/ingest/check-updates` compares the `content_hash` now recorded on the tenant's Neo4j Document nodes at ingestion time against the SHA-256 of current Wiki.js page content in a single UNWIND Cypher query, returning `changed` / `new` / `deleted` page lists (auto-generated entity stubs excluded; documents whose stored hash predates hash tracking are flagged `stored_hash_missing` and count as changed). + - `/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. + ### Changed - **BREAKING: `user` is now required on every tenant-data endpoint** - The implicit `jpmschweitzer` default tenant (`DEFAULT_USER`) has been removed everywhere. All endpoints that read or write tenant data (`/query/*`, `/wiki/*`, `/vector/*`, `/graph/*`, `/ingest/*`, `/volatile/*`, `/documents/*`, `/stats`, `/rag/search`) now reject requests without an explicit, non-empty, non-whitespace `user` (HTTP 422), matching the existing `/maintenance/*` pattern. A shared validator (`require_user` dependency / `RequiredUser` model type) also rejects blank users. The Wiki.js change listener now skips changes whose notification email yields no user instead of attributing them to the production tenant. **Caller coordination required:** tatlock and the Scheduler ingest/prefetch/consolidation tasks must send an explicit `user` on every call — see the deploy checklist. diff --git a/src/clients/qdrant_client.py b/src/clients/qdrant_client.py index 61a20a1..9fa49ec 100644 --- a/src/clients/qdrant_client.py +++ b/src/clients/qdrant_client.py @@ -599,10 +599,13 @@ class QdrantClientWrapper: ) for point in points: - all_points.append({ + entry = { "id": str(point.id), "payload": dict(point.payload) if point.payload else {} - }) + } + if with_vectors: + entry["vector"] = point.vector + all_points.append(entry) if next_offset is None: break diff --git a/src/core/dependencies.py b/src/core/dependencies.py index 75501bf..db43b43 100644 --- a/src/core/dependencies.py +++ b/src/core/dependencies.py @@ -175,6 +175,21 @@ def get_redis_client() -> aioredis.Redis: return client +@lru_cache +def get_job_manager() -> "JobManager": + """ + Get Redis-backed JobManager singleton. + + Returns: + JobManager for background job tracking (connects lazily) + """ + from src.jobs.job_manager import JobManager + settings = get_settings() + manager = JobManager(redis_url=settings.redis_url) + logger.debug(f"Created JobManager: {settings.redis_url}") + return manager + + @lru_cache def get_content_extractor() -> ContentExtractor: """ @@ -368,6 +383,9 @@ PaperlessDep = Annotated[PaperlessClient, Depends(get_paperless_client)] SettingsClientDep = Annotated[SettingsClient, Depends(get_settings_client)] SchedulerDep = Annotated[SchedulerClient, Depends(get_scheduler_client)] +from src.jobs.job_manager import JobManager # noqa: E402 +JobManagerDep = Annotated[JobManager, Depends(get_job_manager)] + # External API provider dependencies WeatherProviderDep = Annotated[OpenMeteoProvider, Depends(get_weather_provider)] NewsProviderDep = Annotated[AggregatedNewsProvider, Depends(get_news_provider)] @@ -527,6 +545,14 @@ async def shutdown_clients(): except Exception as e: logger.error(f"Error closing scheduler client: {e}") + # Close job manager Redis connection + try: + job_manager = get_job_manager() + await job_manager.close() + logger.info("✓ JobManager closed") + except Exception as e: + logger.error(f"Error closing JobManager: {e}") + logger.info("Service clients shutdown complete") diff --git a/src/core/hashing.py b/src/core/hashing.py new file mode 100644 index 0000000..f1e9b4e --- /dev/null +++ b/src/core/hashing.py @@ -0,0 +1,26 @@ +""" +Content hashing helpers. + +A single canonical hash implementation is used everywhere page content is +fingerprinted (Document nodes at ingestion time, /ingest/check-updates +comparisons) so hashes computed at different times are comparable. +""" + +import hashlib + + +def compute_content_hash(content: str) -> str: + """ + Compute the canonical content hash for wiki page content. + + Args: + content: Raw page content (markdown). None-safe: treated as "". + + Returns: + Hex-encoded SHA-256 digest of the UTF-8 encoded content. + + Examples: + >>> compute_content_hash("hello") + '2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824' + """ + return hashlib.sha256((content or "").encode("utf-8")).hexdigest() diff --git a/src/main.py b/src/main.py index 2ab547e..c1f3720 100644 --- a/src/main.py +++ b/src/main.py @@ -11,7 +11,7 @@ Following best practices: from fastapi import FastAPI, HTTPException, Depends, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel +from pydantic import BaseModel, Field from typing import Dict, Any import logging from pathlib import Path @@ -19,8 +19,9 @@ from pathlib import Path from src.config import Settings, get_settings, __version__ from src.core.dependencies import ( verify_api_key, QdrantDep, WikiJSDep, OllamaDep, Neo4jDep, PaperlessDep, - RequiredUserQuery + RequiredUserQuery, JobManagerDep ) +from src.core.multi_tenancy import RequiredUser # Configure logging logging.basicConfig( @@ -228,61 +229,227 @@ async def stats( ) +class CheckUpdatesRequest(BaseModel): + """Request body for /ingest/check-updates.""" + user: RequiredUser = Field( + ..., + description="User identifier (tenant). Required — only this tenant's namespace is compared." + ) + path_prefix: str | None = Field( + default=None, + description="Optional sub-path inside the tenant namespace (e.g. 'technology')" + ) + + @app.post("/ingest/check-updates", tags=["Ingestion"]) async def check_updates( - documents: Dict[str, Any], + request: CheckUpdatesRequest, + neo4j: Neo4jDep = None, + wikijs: WikiJSDep = None, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ - Check which documents need updating based on content hashes. - Used by Scheduler to determine what changed since last sync. + Check which wiki pages need (re-)ingestion based on content hashes. - TODO: Implement update detection: - 1. Query existing documents by path - 2. Compare content hashes - 3. Return list of updates needed + Compares the `content_hash` stored on the tenant's Neo4j Document nodes + (recorded at ingestion time) against the SHA-256 of the current Wiki.js + page content, in a single UNWIND Cypher query. Used by the Scheduler to + determine what changed since the last sync. Read-only. + + Returns per-tenant lists: + - `changed`: page exists in wiki AND graph, but hashes differ (or the + stored hash predates hash tracking — flagged `stored_hash_missing`) + - `new`: wiki page with no Document node yet + - `deleted`: Document node whose wiki page no longer exists """ - return { - "message": "Update checking not yet implemented", - "updates_needed": [], - "up_to_date": [], - "new_documents": [] - } + import time as _time + from src.core.hashing import compute_content_hash + from src.core.multi_tenancy import get_neo4j_user_label, sanitize_user_id + + start_time = _time.time() + user = request.user + tenant_prefix = f"users/{sanitize_user_id(user)}" + if request.path_prefix: + tenant_prefix = f"{tenant_prefix}/{request.path_prefix.strip('/')}" + + try: + pages = await wikijs.list_all_pages(path_prefix=tenant_prefix) + + # Auto-generated entity stubs are intentionally never ingested into + # the graph (see GraphService.update_from_page), so they would show + # up as perpetually "new". Exclude them. + pages = [ + p for p in pages + if not ({"entity-stub", "auto-generated"} & set(p.get("tags") or [])) + ] + + page_hashes = [] + for p in pages: + full_page = await wikijs.get_page(p["id"]) + content = (full_page or {}).get("content", "") + page_hashes.append({ + "page_id": p["id"], + "path": p.get("path", ""), + "title": p.get("title", ""), + "hash": compute_content_hash(content) + }) + + user_doc_label = get_neo4j_user_label(user) + + if page_hashes: + # Single UNWIND query: compare every current page hash against the + # stored Document hash AND collect stale Document nodes whose wiki + # page is gone. + cypher = f""" + UNWIND $pages AS p + OPTIONAL MATCH (d:{user_doc_label}:Document {{page_id: p.page_id}}) + WITH collect({{ + page_id: p.page_id, + path: p.path, + title: p.title, + is_new: d IS NULL, + changed: d IS NOT NULL AND (d.content_hash IS NULL OR d.content_hash <> p.hash), + stored_hash_missing: d IS NOT NULL AND d.content_hash IS NULL + }}) AS checked, + collect(p.page_id) AS current_ids + OPTIONAL MATCH (stale:{user_doc_label}:Document) + WHERE stale.page_id IS NOT NULL AND NOT stale.page_id IN current_ids + RETURN checked, + collect(CASE WHEN stale IS NULL THEN NULL ELSE {{ + page_id: stale.page_id, path: stale.path, title: stale.title + }} END) AS deleted + """ + rows = await neo4j.execute_query(cypher, {"pages": page_hashes}) + checked = rows[0]["checked"] if rows else [] + deleted = rows[0]["deleted"] if rows else [] + else: + # No wiki pages under the prefix: every Document node is stale. + cypher = f""" + MATCH (stale:{user_doc_label}:Document) + WHERE stale.page_id IS NOT NULL + RETURN collect({{page_id: stale.page_id, path: stale.path, title: stale.title}}) AS deleted + """ + rows = await neo4j.execute_query(cypher, {}) + checked = [] + deleted = rows[0]["deleted"] if rows else [] + + # Deleted detection is namespace-wide only for full-tenant scans; a + # sub-path scan must not flag documents outside its prefix. + if request.path_prefix: + deleted = [ + d for d in deleted + if str(d.get("path", "")).lstrip("/").startswith(tenant_prefix) + ] + + new_pages = [c for c in checked if c["is_new"]] + changed_pages = [c for c in checked if c["changed"]] + up_to_date = len(checked) - len(new_pages) - len(changed_pages) + + duration_ms = (_time.time() - start_time) * 1000 + + return { + "user": user, + "path_prefix": tenant_prefix, + "total_wiki_pages": len(page_hashes), + "changed": [ + {k: c[k] for k in ("page_id", "path", "title", "stored_hash_missing")} + for c in changed_pages + ], + "new": [ + {k: c[k] for k in ("page_id", "path", "title")} for c in new_pages + ], + "deleted": deleted, + "counts": { + "changed": len(changed_pages), + "new": len(new_pages), + "deleted": len(deleted), + "up_to_date": up_to_date + }, + "duration_ms": duration_ms + } + + except Exception as e: + logger.error(f"check-updates failed for {user}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Update check failed") -@app.get("/ingest/status/{document_id}", tags=["Ingestion"]) +@app.get("/ingest/status/{job_id}", tags=["Ingestion"]) async def get_ingestion_status( - document_id: str, + job_id: str, + user: RequiredUserQuery, + job_manager: JobManagerDep = None, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ - Get processing status for a document. + Get processing status for an ingestion job. - TODO: Implement status tracking + Backed by the Redis job store (`library:job:{job_id}`, 24h TTL). Job IDs + are returned by /ingest/page, /ingest/batch and /ingest/all. Jobs are + tenant-scoped: requesting another tenant's job returns 404. """ - return { - "message": "Status tracking not yet implemented", - "document_id": document_id, - "status": "unknown" - } + job = await job_manager.get_job(job_id) + if not job or job.get("user") != user: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + return job @app.get("/ingest/repo-status/{repository}", tags=["Ingestion"]) async def get_repo_status( repository: str, + user: RequiredUserQuery, + neo4j: Neo4jDep = None, + wikijs: WikiJSDep = None, + job_manager: JobManagerDep = None, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ - Get indexing status for an entire repository. + Get indexing status for a repository (a sub-path of the tenant namespace). - TODO: Implement repository-level statistics + `repository` is resolved as `users/{tenant}/{repository}`; use `_all` for + the whole tenant namespace. Reports how many wiki pages exist under the + path, how many have graph Document nodes (i.e. are indexed), and the + tenant's recent job statistics from the Redis job store. """ - return { - "message": "Repository status not yet implemented", - "repository": repository, - "total_documents": 0, - "indexed_documents": 0 - } + import time as _time + from src.core.multi_tenancy import get_neo4j_user_label, sanitize_user_id + + start_time = _time.time() + tenant_root = f"users/{sanitize_user_id(user)}" + prefix = tenant_root if repository in ("_all", "all", "") else f"{tenant_root}/{repository.strip('/')}" + + try: + pages = await wikijs.list_all_pages(path_prefix=prefix) + page_ids = [p["id"] for p in pages if p.get("id")] + + indexed = 0 + if page_ids: + user_doc_label = get_neo4j_user_label(user) + rows = await neo4j.execute_query( + f""" + MATCH (d:{user_doc_label}:Document) + WHERE d.page_id IN $page_ids + RETURN count(DISTINCT d.page_id) AS indexed + """, + {"page_ids": page_ids} + ) + indexed = rows[0]["indexed"] if rows else 0 + + job_stats = await job_manager.get_job_stats(user=user) + + return { + "repository": repository, + "user": user, + "path_prefix": prefix, + "total_documents": len(page_ids), + "indexed_documents": indexed, + "unindexed_documents": len(page_ids) - indexed, + "jobs": job_stats, + "duration_ms": (_time.time() - start_time) * 1000 + } + + except Exception as e: + logger.error(f"repo-status failed for {user}/{repository}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Repository status failed") # Query endpoints @@ -373,35 +540,58 @@ async def graph_query( # Deduplication endpoints +class DeduplicateCheckRequest(BaseModel): + """Request body for /deduplicate/check.""" + user: RequiredUser = Field( + ..., + description="User identifier (tenant). Required — only this tenant's collection is scanned." + ) + similarity_threshold: float = Field( + default=0.9, ge=0.5, le=1.0, + description="Minimum cosine similarity for a chunk pair to count as duplicate" + ) + max_pairs: int = Field(default=100, ge=1, le=500, description="Maximum page pairs returned") + + @app.post("/deduplicate/check", tags=["Deduplication"]) async def check_duplicates( - request: Dict[str, Any], + request: DeduplicateCheckRequest, + qdrant_client: QdrantDep = None, + wiki_client: WikiJSDep = None, + ollama_client: OllamaDep = None, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ - Check for duplicate or highly similar documents. - Uses vector similarity and graph analysis. + Check for duplicate or highly similar wiki pages (tenant-scoped, read-only). - Expected fields: - - document_id: str - - similarity_threshold: float (default 0.85) - - TODO: Implement deduplication: - 1. Get document embedding from Qdrant - 2. Find similar vectors above threshold - 3. Check graph relationships - 4. Return candidates with similarity scores + Scans the tenant's own Qdrant collection: every wiki chunk vector is + queried against the same collection, and chunk pairs from different + pages scoring above the threshold (default 0.9 cosine) are grouped per + page pair with the best similarity and matching chunk-pair count. """ - document_id = request.get("document_id") - threshold = request.get("similarity_threshold", 0.85) + import time as _time + from src.services.vector_service import VectorService - return { - "message": "Deduplication not yet implemented", - "document_id": document_id, - "threshold": threshold, - "duplicates": [], - "suggestions": None - } + start_time = _time.time() + vector_service = VectorService(qdrant_client, wiki_client, ollama_client) + + try: + scan = await vector_service.find_duplicate_pairs( + user=request.user, + similarity_threshold=request.similarity_threshold, + max_pairs=request.max_pairs + ) + return { + "user": request.user, + "similarity_threshold": request.similarity_threshold, + "chunks_scanned": scan["chunks_scanned"], + "duplicate_groups": scan["duplicate_groups"], + "duplicate_group_count": len(scan["duplicate_groups"]), + "duration_ms": (_time.time() - start_time) * 1000 + } + except Exception as e: + logger.error(f"Deduplication check failed for {request.user}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Deduplication check failed") # Application lifecycle diff --git a/src/models/ingestion.py b/src/models/ingestion.py index c7b2b76..7445a3e 100644 --- a/src/models/ingestion.py +++ b/src/models/ingestion.py @@ -46,6 +46,10 @@ class IngestionResult(BaseModel): graph_entities_extracted: int = 0 graph_relationships_created: int = 0 processing_time_ms: float + job_id: Optional[str] = Field( + default=None, + description="Redis job-tracking ID (query via GET /ingest/status/{job_id})" + ) class BatchIngestionResult(BaseModel): @@ -55,6 +59,10 @@ class BatchIngestionResult(BaseModel): failed: int results: List[IngestionResult] total_processing_time_ms: float + job_id: Optional[str] = Field( + default=None, + description="Redis job-tracking ID (query via GET /ingest/status/{job_id})" + ) class IngestionStatus(BaseModel): diff --git a/src/routers/ingestion.py b/src/routers/ingestion.py index b18df36..ed07e5d 100644 --- a/src/routers/ingestion.py +++ b/src/routers/ingestion.py @@ -5,6 +5,7 @@ Endpoints for ingesting wiki pages into the knowledge base (vectors + graph). """ from fastapi import APIRouter, Depends, HTTPException, Query from typing import Optional +import logging from src.services.ingestion_service import IngestionService from src.models.ingestion import ( @@ -13,15 +14,57 @@ from src.models.ingestion import ( BatchIngestionRequest, BatchIngestionResult ) -from src.core.dependencies import get_ingestion_service, verify_api_key, RequiredUserQuery +from src.core.dependencies import ( + get_ingestion_service, verify_api_key, RequiredUserQuery, JobManagerDep +) +from src.jobs.job_manager import JobManager, JobStatus, JobType + +logger = logging.getLogger(__name__) router = APIRouter(prefix="/ingest", tags=["Document Ingestion"]) +async def _track_job( + job_manager: JobManager, + job_type: JobType, + user: str, + parameters: dict +) -> Optional[str]: + """Create a Redis job record; never fail the request over job tracking.""" + try: + return await job_manager.create_job(job_type, user, parameters) + except Exception as e: + logger.warning(f"Job tracking unavailable ({job_type.value}): {e}") + return None + + +async def _finish_job( + job_manager: JobManager, + job_id: Optional[str], + success: bool, + result: dict, + error: Optional[str] = None +) -> None: + """Mark a tracked job completed/failed; never fail the request.""" + if not job_id: + return + try: + await job_manager.update_job_status( + job_id, + JobStatus.COMPLETED if success else JobStatus.FAILED, + progress=100, + result=result, + error=error + ) + except Exception as e: + logger.warning(f"Job tracking update failed for {job_id}: {e}") + + @router.post("/page", response_model=IngestionResult) async def ingest_page( request: IngestionRequest, ingestion: IngestionService = Depends(get_ingestion_service), + job_manager: JobManagerDep = None, api_key: str = Depends(verify_api_key) ): """ @@ -57,6 +100,11 @@ async def ingest_page( }' ``` """ + job_id = await _track_job( + job_manager, JobType.DOCUMENT_INGESTION, request.user, + {"page_id": request.page_id, "force_refresh": request.force_refresh} + ) + result = await ingestion.ingest_page( page_id=request.page_id, user=request.user, @@ -64,6 +112,12 @@ async def ingest_page( skip_vectors=request.skip_vectors, skip_graph=request.skip_graph ) + result.job_id = job_id + + await _finish_job( + job_manager, job_id, result.success, + result=result.model_dump(mode="json"), error=result.error + ) if not result.success: raise HTTPException( @@ -78,6 +132,7 @@ async def ingest_page( async def ingest_batch( request: BatchIngestionRequest, ingestion: IngestionService = Depends(get_ingestion_service), + job_manager: JobManagerDep = None, api_key: str = Depends(verify_api_key) ): """ @@ -109,6 +164,11 @@ async def ingest_batch( }' ``` """ + job_id = await _track_job( + job_manager, JobType.BATCH_INGESTION, request.user, + {"page_ids": request.page_ids, "force_refresh": request.force_refresh} + ) + result = await ingestion.ingest_batch( page_ids=request.page_ids, user=request.user, @@ -117,6 +177,16 @@ async def ingest_batch( skip_graph=request.skip_graph, max_concurrent=request.max_concurrent ) + result.job_id = job_id + + await _finish_job( + job_manager, job_id, result.failed == 0, + result={ + "total_pages": result.total_pages, + "successful": result.successful, + "failed": result.failed + } + ) return result @@ -128,6 +198,7 @@ async def ingest_all_pages( force_refresh: bool = Query(False, description="Force re-ingestion of all pages"), max_concurrent: int = Query(3, ge=1, le=10, description="Maximum concurrent ingestion tasks"), ingestion: IngestionService = Depends(get_ingestion_service), + job_manager: JobManagerDep = None, api_key: str = Depends(verify_api_key) ): """ @@ -159,6 +230,11 @@ async def ingest_all_pages( -H "Authorization: Bearer $API_KEY" ``` """ + job_id = await _track_job( + job_manager, JobType.BATCH_INGESTION, user, + {"path_prefix": path_prefix, "force_refresh": force_refresh, "scope": "all"} + ) + try: result = await ingestion.ingest_all_pages( user=user, @@ -167,6 +243,17 @@ async def ingest_all_pages( max_concurrent=max_concurrent ) except ValueError as e: + await _finish_job(job_manager, job_id, False, result={}, error=str(e)) raise HTTPException(status_code=400, detail=str(e)) + result.job_id = job_id + await _finish_job( + job_manager, job_id, result.failed == 0, + result={ + "total_pages": result.total_pages, + "successful": result.successful, + "failed": result.failed + } + ) + return result diff --git a/src/services/graph_service.py b/src/services/graph_service.py index 06dfd32..29e98c5 100644 --- a/src/services/graph_service.py +++ b/src/services/graph_service.py @@ -452,14 +452,20 @@ class GraphService: user_base_label = get_neo4j_user_base_label(user) # For entities user_doc_label = get_neo4j_user_label(user) # For documents - # Create/update Document node + # Create/update Document node. + # content_hash records the fingerprint of the ingested content so + # /ingest/check-updates can detect changed pages without re-reading + # the graph's source content. + from src.core.hashing import compute_content_hash + doc_query = f""" MERGE (d:{user_doc_label}:Document {{page_id: $page_id}}) SET d.title = $title, d.path = $path, d.tags = $tags, d.updated_at = datetime(), - d.content_length = $content_length + d.content_length = $content_length, + d.content_hash = $content_hash RETURN d """ @@ -468,7 +474,8 @@ class GraphService: "title": page.get("title"), "path": page.get("path"), "tags": tags, - "content_length": len(content) + "content_length": len(content), + "content_hash": compute_content_hash(content) }) nodes_created = 1 # Document node diff --git a/src/services/vector_service.py b/src/services/vector_service.py index 65a017f..f189d13 100644 --- a/src/services/vector_service.py +++ b/src/services/vector_service.py @@ -543,6 +543,129 @@ class VectorService: logger.error(f"Failed to purge chunks: {e}", exc_info=True) return 0 + async def find_duplicate_pairs( + self, + user: str, + similarity_threshold: float = 0.9, + max_chunks_scanned: int = 2000, + max_pairs: int = 100 + ) -> Dict[str, Any]: + """ + Tenant-scoped similarity scan for near-duplicate wiki pages. + + Scrolls the tenant's own Qdrant collection (never another tenant's), + then queries each chunk's vector against the same collection. Chunk + pairs from DIFFERENT pages scoring above the threshold are grouped + per page pair with the best score and the number of matching chunk + pairs. Read-only: nothing is modified. + + Args: + user: Tenant user identifier + similarity_threshold: Minimum cosine similarity (default 0.9) + max_chunks_scanned: Safety cap on chunks used as probes + max_pairs: Maximum page pairs returned (highest score first) + + Returns: + { + "chunks_scanned": int, + "duplicate_groups": [ + { + "pages": [{page_id, path, title}, {page_id, path, title}], + "max_similarity": float, + "matching_chunk_pairs": int + }, ... + ] + } + """ + collection_name = get_qdrant_collection_name(user) + + exists = await self.qdrant.collection_exists(collection_name) + if not exists: + return {"chunks_scanned": 0, "duplicate_groups": []} + + points = await self.qdrant.scroll_all_points( + collection_name=collection_name, + batch_size=100, + with_payload=True, + with_vectors=True + ) + + # Only wiki chunks participate (documents have their own dedup story) + wiki_points = [ + p for p in points + if p.get("vector") is not None + and (p.get("payload") or {}).get("doc_type", "wiki") == "wiki" + and (p.get("payload") or {}).get("page_id") + ][:max_chunks_scanned] + + page_meta: Dict[int, Dict[str, Any]] = {} + pair_stats: Dict[tuple, Dict[str, Any]] = {} + seen_chunk_pairs = set() + + for point in wiki_points: + payload = point.get("payload") or {} + page_id = payload.get("page_id") + page_meta.setdefault(page_id, { + "page_id": page_id, + "path": payload.get("page_path", ""), + "title": payload.get("page_title", "") + }) + + hits = await self.qdrant.search_vectors( + collection_name=collection_name, + query_vector=point["vector"], + limit=10, + score_threshold=similarity_threshold + ) + + for hit in hits: + hit_payload = hit.get("payload") or {} + hit_page_id = hit_payload.get("page_id") + if not hit_page_id or hit_page_id == page_id: + continue + if hit_payload.get("doc_type", "wiki") != "wiki": + continue + + # Deduplicate the A->B / B->A chunk pair directions + chunk_pair = tuple(sorted((point["id"], hit["id"]))) + if chunk_pair in seen_chunk_pairs: + continue + seen_chunk_pairs.add(chunk_pair) + + page_meta.setdefault(hit_page_id, { + "page_id": hit_page_id, + "path": hit_payload.get("page_path", ""), + "title": hit_payload.get("page_title", "") + }) + + page_pair = tuple(sorted((page_id, hit_page_id))) + stats = pair_stats.setdefault(page_pair, { + "max_similarity": 0.0, + "matching_chunk_pairs": 0 + }) + stats["max_similarity"] = max(stats["max_similarity"], hit["score"]) + stats["matching_chunk_pairs"] += 1 + + groups = [ + { + "pages": [page_meta[a], page_meta[b]], + "max_similarity": stats["max_similarity"], + "matching_chunk_pairs": stats["matching_chunk_pairs"] + } + for (a, b), stats in pair_stats.items() + ] + groups.sort(key=lambda g: g["max_similarity"], reverse=True) + + logger.info( + f"Duplicate scan for {user}: {len(wiki_points)} chunks scanned, " + f"{len(groups)} page pairs above {similarity_threshold}" + ) + + return { + "chunks_scanned": len(wiki_points), + "duplicate_groups": groups[:max_pairs] + } + def find_chunks_without_graph_nodes( self, chunk_references: List[Dict[str, Any]], diff --git a/tests/test_ingest_endpoints.py b/tests/test_ingest_endpoints.py new file mode 100644 index 0000000..6d682d5 --- /dev/null +++ b/tests/test_ingest_endpoints.py @@ -0,0 +1,320 @@ +""" +Offline unit tests for the implemented /ingest and /deduplicate endpoints. + +Covers (Phase C item 1): +- /ingest/check-updates: content-hash comparison classification +- /ingest/status/{job_id}: Redis job-manager backing + tenant scoping +- /ingest/repo-status/{repository}: wiki vs graph counts + job stats +- /deduplicate/check: tenant-scoped similarity scan grouping + +All external clients are mocked - no shared services are contacted. +""" + +import pytest +from unittest.mock import AsyncMock + +from fastapi import HTTPException + +from src.core.hashing import compute_content_hash +from src.main import ( + check_updates, + check_duplicates, + get_ingestion_status, + get_repo_status, + CheckUpdatesRequest, + DeduplicateCheckRequest, +) +from src.services.vector_service import VectorService + +TEST_USER = "llm_tester" + + +# --------------------------------------------------------------------------- +# /ingest/check-updates +# --------------------------------------------------------------------------- + + +class TestCheckUpdates: + @pytest.mark.asyncio + async def test_classifies_changed_new_and_deleted(self): + """Hash mismatch -> changed, no node -> new, stale node -> deleted.""" + wikijs = AsyncMock() + wikijs.list_all_pages = AsyncMock(return_value=[ + {"id": 1, "path": f"users/{TEST_USER}/a", "title": "A", "tags": []}, + {"id": 2, "path": f"users/{TEST_USER}/b", "title": "B", "tags": []}, + ]) + wikijs.get_page = AsyncMock(side_effect=[ + {"id": 1, "content": "content-a"}, + {"id": 2, "content": "content-b"}, + ]) + + neo4j = AsyncMock() + + async def fake_query(cypher, params): + # The endpoint sends one UNWIND query with page hashes + pages = params["pages"] + assert pages[0]["hash"] == compute_content_hash("content-a") + return [{ + "checked": [ + { # page 1: node exists, hash matches -> up to date + "page_id": 1, "path": pages[0]["path"], "title": "A", + "is_new": False, "changed": False, + "stored_hash_missing": False, + }, + { # page 2: node exists, hash differs -> changed + "page_id": 2, "path": pages[1]["path"], "title": "B", + "is_new": False, "changed": True, + "stored_hash_missing": False, + }, + ], + "deleted": [ + {"page_id": 99, "path": f"users/{TEST_USER}/gone", "title": "Gone"}, + ], + }] + + neo4j.execute_query = AsyncMock(side_effect=fake_query) + + result = await check_updates( + request=CheckUpdatesRequest(user=TEST_USER), + neo4j=neo4j, + wikijs=wikijs, + api_key="", + ) + + assert result["counts"] == { + "changed": 1, "new": 0, "deleted": 1, "up_to_date": 1 + } + assert result["changed"][0]["page_id"] == 2 + assert result["deleted"][0]["page_id"] == 99 + assert result["duration_ms"] >= 0 + + @pytest.mark.asyncio + async def test_new_page_detected_and_stub_pages_excluded(self): + """Pages without Document nodes are new; entity stubs are skipped.""" + wikijs = AsyncMock() + wikijs.list_all_pages = AsyncMock(return_value=[ + {"id": 5, "path": f"users/{TEST_USER}/fresh", "title": "Fresh", "tags": []}, + {"id": 6, "path": f"users/{TEST_USER}/stub", "title": "Stub", + "tags": ["entity-stub"]}, + ]) + wikijs.get_page = AsyncMock(return_value={"id": 5, "content": "x"}) + + neo4j = AsyncMock() + neo4j.execute_query = AsyncMock(return_value=[{ + "checked": [{ + "page_id": 5, "path": f"users/{TEST_USER}/fresh", "title": "Fresh", + "is_new": True, "changed": False, "stored_hash_missing": False, + }], + "deleted": [], + }]) + + result = await check_updates( + request=CheckUpdatesRequest(user=TEST_USER), + neo4j=neo4j, + wikijs=wikijs, + api_key="", + ) + + # Only page 5 was hashed (stub excluded -> get_page called once) + assert wikijs.get_page.await_count == 1 + assert result["counts"]["new"] == 1 + assert result["new"][0]["page_id"] == 5 + + @pytest.mark.asyncio + async def test_empty_wiki_reports_all_documents_deleted(self): + """With no wiki pages, every Document node is reported deleted.""" + wikijs = AsyncMock() + wikijs.list_all_pages = AsyncMock(return_value=[]) + + neo4j = AsyncMock() + neo4j.execute_query = AsyncMock(return_value=[{ + "deleted": [{"page_id": 7, "path": f"users/{TEST_USER}/x", "title": "X"}], + }]) + + result = await check_updates( + request=CheckUpdatesRequest(user=TEST_USER), + neo4j=neo4j, + wikijs=wikijs, + api_key="", + ) + + assert result["counts"] == { + "changed": 0, "new": 0, "deleted": 1, "up_to_date": 0 + } + + def test_requires_user(self): + """Empty user is rejected by the request model (Phase B rule).""" + with pytest.raises(Exception): + CheckUpdatesRequest(user=" ") + + +# --------------------------------------------------------------------------- +# /ingest/status/{job_id} +# --------------------------------------------------------------------------- + + +class TestIngestionStatus: + @pytest.mark.asyncio + async def test_returns_job_from_job_manager(self): + job_manager = AsyncMock() + job_manager.get_job = AsyncMock(return_value={ + "job_id": "abc", "user": TEST_USER, "status": "completed", + }) + + job = await get_ingestion_status( + job_id="abc", user=TEST_USER, job_manager=job_manager, api_key="" + ) + assert job["status"] == "completed" + job_manager.get_job.assert_awaited_once_with("abc") + + @pytest.mark.asyncio + async def test_unknown_job_404(self): + job_manager = AsyncMock() + job_manager.get_job = AsyncMock(return_value=None) + + with pytest.raises(HTTPException) as exc: + await get_ingestion_status( + job_id="missing", user=TEST_USER, job_manager=job_manager, api_key="" + ) + assert exc.value.status_code == 404 + + @pytest.mark.asyncio + async def test_other_tenants_job_is_hidden(self): + """Tenant scoping: another user's job looks like a 404.""" + job_manager = AsyncMock() + job_manager.get_job = AsyncMock(return_value={ + "job_id": "abc", "user": "someone_else", "status": "completed", + }) + + with pytest.raises(HTTPException) as exc: + await get_ingestion_status( + job_id="abc", user=TEST_USER, job_manager=job_manager, api_key="" + ) + assert exc.value.status_code == 404 + + +# --------------------------------------------------------------------------- +# /ingest/repo-status/{repository} +# --------------------------------------------------------------------------- + + +class TestRepoStatus: + @pytest.mark.asyncio + async def test_counts_indexed_vs_total(self): + wikijs = AsyncMock() + wikijs.list_all_pages = AsyncMock(return_value=[ + {"id": 1, "path": f"users/{TEST_USER}/tech/a"}, + {"id": 2, "path": f"users/{TEST_USER}/tech/b"}, + {"id": 3, "path": f"users/{TEST_USER}/tech/c"}, + ]) + neo4j = AsyncMock() + neo4j.execute_query = AsyncMock(return_value=[{"indexed": 2}]) + job_manager = AsyncMock() + job_manager.get_job_stats = AsyncMock(return_value={"total": 4, "completed": 4}) + + result = await get_repo_status( + repository="tech", user=TEST_USER, + neo4j=neo4j, wikijs=wikijs, job_manager=job_manager, api_key="" + ) + + assert result["total_documents"] == 3 + assert result["indexed_documents"] == 2 + assert result["unindexed_documents"] == 1 + assert result["jobs"]["total"] == 4 + assert result["path_prefix"] == f"users/{TEST_USER}/tech" + # Wiki listing was scoped to the tenant namespace + wikijs.list_all_pages.assert_awaited_once_with( + path_prefix=f"users/{TEST_USER}/tech" + ) + + +# --------------------------------------------------------------------------- +# /deduplicate/check + VectorService.find_duplicate_pairs +# --------------------------------------------------------------------------- + + +def _make_vector_service(points, search_results_by_id): + qdrant = AsyncMock() + qdrant.collection_exists = AsyncMock(return_value=True) + qdrant.scroll_all_points = AsyncMock(return_value=points) + + async def fake_search(collection_name, query_vector, limit, score_threshold): + # Route on the probe vector's marker value + return search_results_by_id.get(query_vector[0], []) + + qdrant.search_vectors = AsyncMock(side_effect=fake_search) + return VectorService(qdrant, AsyncMock(), AsyncMock()), qdrant + + +class TestDeduplicateCheck: + @pytest.mark.asyncio + async def test_groups_pairs_by_page_and_dedupes_directions(self): + points = [ + {"id": "c1", "vector": [1.0], + "payload": {"page_id": 10, "page_path": "users/llm_tester/a", + "page_title": "A", "doc_type": "wiki"}}, + {"id": "c2", "vector": [2.0], + "payload": {"page_id": 20, "page_path": "users/llm_tester/b", + "page_title": "B", "doc_type": "wiki"}}, + ] + search_results = { + 1.0: [ # c1 finds c2 (cross-page) and itself (same page - ignored) + {"id": "c1", "score": 1.0, "payload": points[0]["payload"]}, + {"id": "c2", "score": 0.95, "payload": points[1]["payload"]}, + ], + 2.0: [ # c2 finds c1 - reverse direction must NOT double count + {"id": "c1", "score": 0.95, "payload": points[0]["payload"]}, + ], + } + service, _ = _make_vector_service(points, search_results) + + scan = await service.find_duplicate_pairs(TEST_USER, similarity_threshold=0.9) + + assert scan["chunks_scanned"] == 2 + assert len(scan["duplicate_groups"]) == 1 + group = scan["duplicate_groups"][0] + assert group["max_similarity"] == 0.95 + assert group["matching_chunk_pairs"] == 1 + assert {p["page_id"] for p in group["pages"]} == {10, 20} + + @pytest.mark.asyncio + async def test_missing_collection_returns_empty(self): + qdrant = AsyncMock() + qdrant.collection_exists = AsyncMock(return_value=False) + service = VectorService(qdrant, AsyncMock(), AsyncMock()) + + scan = await service.find_duplicate_pairs(TEST_USER) + assert scan == {"chunks_scanned": 0, "duplicate_groups": []} + qdrant.scroll_all_points.assert_not_awaited() + + @pytest.mark.asyncio + async def test_endpoint_shape(self): + service_qdrant = AsyncMock() + service_qdrant.collection_exists = AsyncMock(return_value=False) + + result = await check_duplicates( + request=DeduplicateCheckRequest(user=TEST_USER), + qdrant_client=service_qdrant, + wiki_client=AsyncMock(), + ollama_client=AsyncMock(), + api_key="", + ) + assert result["user"] == TEST_USER + assert result["similarity_threshold"] == 0.9 + assert result["duplicate_groups"] == [] + assert result["duplicate_group_count"] == 0 + + def test_requires_user(self): + with pytest.raises(Exception): + DeduplicateCheckRequest(user="") + + +# --------------------------------------------------------------------------- +# content hash canonicality +# --------------------------------------------------------------------------- + + +def test_content_hash_is_stable_and_none_safe(): + assert compute_content_hash("abc") == compute_content_hash("abc") + assert compute_content_hash("abc") != compute_content_hash("abd") + assert compute_content_hash(None) == compute_content_hash("")