feat: implement check-updates, job-backed ingest status, and dedup scan

Replace the four stub endpoints with real implementations, all requiring
an explicit tenant user (Phase B rule):

- /ingest/check-updates: GraphService now records a SHA-256 content_hash
  on every Document node at ingestion time; the endpoint compares those
  stored hashes against current Wiki.js page content in one UNWIND Cypher
  query per tenant and returns changed/new/deleted page lists (entity-stub
  pages excluded, pre-hash-tracking documents flagged stored_hash_missing).
- /ingest/status/{job_id}: backed by the Redis JobManager; jobs are
  tenant-scoped (foreign jobs 404). /ingest/page, /ingest/batch and
  /ingest/all now create job records and return job_id.
- /ingest/repo-status/{repository}: wiki page count vs indexed Document
  nodes under users/{tenant}/{repository} plus tenant job stats.
- /deduplicate/check: tenant-scoped Qdrant similarity scan; chunk pairs
  above ~0.9 cosine from different pages grouped per page pair with best
  score and page references (read-only).

Supporting changes: get_job_manager dependency (+ shutdown close),
scroll_all_points can return vectors, VectorService.find_duplicate_pairs,
src/core/hashing.compute_content_hash. 13 new offline unit tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

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

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")
 
 

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()

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

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):

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

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

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]],